1,首先引入配置文件org.springframework.amqp,如下:
<dependency><groupId>org.springframework.amqp</groupId><artifactId>spring-rabbit</artifactId><version>1.7.1.RELEASE</version></dependency>
2,准备工作:安装好rabbitmq,并在项目中增加配置文件 rabbit.properties 内容如下:
rmq.ip=192.188.113.114
rmq.port=5672
rmq.producer.num=20
rmq.manager.user=admin
rmq.manager.password=admin
3,rabbitmq属性介绍:
概念解释:
Brocker:消息队列服务器实体。Exchange:消息交换机,指定消息按什么规则,路由到哪个队列。Queue:消息队列,每个消息都会被投入到一个或者多个队列里。Binding:绑定,它的作用是把exchange和queue按照路由规则binding起来。Routing Key:路由关键字,exchange根据这个关键字进行消息投递。Virtual Host: 虚拟主机,一个broker里可以开设多个vhost,用作不用用户的权限分离。每个virtual host本质上都是一个RabbitMQ Server(但是一个server中可以有多个virtual host),拥有它自己若干的个Exchange、Queue和bings rule等等。其实这是一个虚拟概念,类似于权限控制组。Virtual Host是权限控制的最小粒度。Producer:消息生产者,就是投递消息的程序。Consumer:消息消费者,就是接受消息的程序。Connection: 就是一个TCP的连接。Producer和Consumer都是通过TCP连接到RabbitMQ Server的。接下来的实践案例中我们就可以看到,producer和consumer与exchange的通信的前提是先建立TCP连接。仅仅创建了TCP连接,producer和consumer与exchange还是不能通信的。我们还需要为每一个Connection创建Channel。Channel: 它是建立在上述TCP连接之上的虚拟连接。数据传输都是在Channel中进行的。AMQP协议规定只有通过Channel才能执行AMQP的命令。一个Connection可以包含多个Channel。有人要问了,为什么要使用Channel呢,直接用TCP连接不就好了么?对于一个消息服务器来说,它的任务是处理海量的消息,当有很多线程需要从RabbitMQ中消费消息或者生产消息,那么必须建立很多个connection,也就是许多个TCP连接。然而对于操作系统而言,建立和关闭TCP连接是非常昂贵的开销,而且TCP的连接数也有限制,频繁的建立关闭TCP连接对于系统的性能有很大的影响,如果遇到高峰,性能瓶颈也随之显现。RabbitMQ采用类似NIO的做法,选择TCP连接服用,不仅可以减少性能开销,同时也便于管理。在TCP连接中建立Channel是没有上述代价的,可以复用TCP连接。对于Producer或者Consumer来说,可以并发的使用多个Channel进行Publish或者Receive。有实验表明,在Channel中,1秒可以Publish10K的数据包。对于普通的Consumer或者Producer来说,这已经足够了。除非有非常大的流量时,一个connection可能会产生性能瓶颈,此时就需要开辟多个connection。
消息队列的使用过程大概如下:
消息接收
客户端连接到消息队列服务器,打开一个channel。
客户端声明一个exchange,并设置相关属性。
客户端声明一个queue,并设置相关属性。
客户端使用routing key,在exchange和queue之间建立好绑定关系。
消息发布
客户端投递消息到exchange。
exchange接收到消息后,就根据消息的key和已经设置的binding,进行消息路由,将消息投递到一个或多个队列里。
AMQP 里主要要说两个组件:
Exchange 和 Queue
绿色的X就是Exchange ,红色的是Queue,这两者都在Server端,又称作Broker
这部分是RabbitMQ实现的,而蓝色的则是客户端,通常有Producer和Consumer两种类型。
4,配置spring-rabbitmq.xml,内容如下:
<!-- 公共部分 --> <!-- 创建连接类 连接安装好的 rabbitmq --> <bean id="connectionFactory" class="org.springframework.amqp.rabbit.connection.CachingConnectionFactory"><constructor-arg value="localhost" /> <!-- username,访问RabbitMQ服务器的账户,默认是guest --><property name="username" value="${rmq.manager.user}" /><!-- username,访问RabbitMQ服务器的密码,默认是guest --> <property name="password" value="${rmq.manager.password}" /><!-- host,RabbitMQ服务器地址,默认值"localhost" --> <property name="host" value="${rmq.ip}" /> <!-- port,RabbitMQ服务端口,默认值为5672 --><property name="port" value="${rmq.port}" /><!-- channel-cache-size,channel的缓存数量,默认值为25 --><property name="channel-cache-size" value="50" /><!-- cache-mode,缓存连接模式,默认值为CHANNEL(单个connection连接,连接之后关闭,自动销毁) --><property name="cache-mode" value="CHANNEL" /> </bean> <!--或者这样配置,connection-factory元素实际就是注册一个org.springframework.amqp.rabbit.connection.CachingConnectionFactory实例 <rabbit:connection-factory id="connectionFactory" host="${rmq.ip}" port="${rmq.port}" username="${rmq.manager.user}" password="${rmq.manager.password}" />--> <rabbit:admin connection-factory="connectionFactory"/><!--定义消息队列,durable:是否持久化,如果想在RabbitMQ退出或崩溃的时候,不会失去所有的queue和消息,需要同时标志队列(queue)和交换机(exchange)是持久化的,即rabbit:queue标签和rabbit:direct-exchange中的durable=true,而消息(message)默认是持久化的可以看类org.springframework.amqp.core.MessageProperties中的属性public static final MessageDeliveryMode DEFAULT_DELIVERY_MODE = MessageDeliveryMode.PERSISTENT;exclusive: 仅创建者可以使用的私有队列,断开后自动删除;auto_delete: 当所有消费客户端连接断开后,是否自动删除队列 --> <rabbit:queue name="spittle.alert.queue.1" id="queue_1" durable="true" auto-delete="false" exclusive="false" /> <rabbit:queue name="spittle.alert.queue.2" id="queue_2" durable="true" auto-delete="false" exclusive="false" /> <rabbit:queue name="spittle.alert.queue.3" id="queue_3" durable="true" auto-delete="false" exclusive="false" /><!--绑定队列,rabbitmq的exchangeType常用的三种模式:direct,fanout,topic三种,我们用direct模式,即rabbit:direct-exchange标签,Direct交换器很简单,如果是Direct类型,就会将消息中的RoutingKey与该Exchange关联的所有Binding中的BindingKey进行比较,如果相等,则发送到该Binding对应的Queue中。有一个需要注意的地方:如果找不到指定的exchange,就会报错。但routing key找不到的话,不会报错,这条消息会直接丢失,所以此处要小心,auto-delete:自动删除,如果为Yes,则该交换机所有队列queue删除后,自动删除交换机,默认为false --> <rabbit:direct-exchange id="spittle.fanout" name="spittle.fanout" durable="true" auto-delete="false"><rabbit:bindings><rabbit:binding queue="spittle.alert.queue.1" key="{alert.queue.1}"></rabbit:binding><rabbit:binding queue="spittle.alert.queue.2" key="{alert.queue.2}"></rabbit:binding><rabbit:binding queue="spittle.alert.queue.3" key="{alert.queue.3}"></rabbit:binding></rabbit:bindings> </rabbit:fanout-exchange><!-- 生产者部分 --> <!-- 发送消息的producer类,也就是生产者 --> <bean id="msgProducer" class="com.asdf.sdf.ClassA"><!-- value中的值就是producer中的的routingKey,也就是队列名称,它与上面的rabbit:bindings标签中的key必须相同 --><property name="queueName" value="{alert.queue.1}"/> </bean><!-- spring amqp默认的是jackson 的一个插件,目的将生产者生产的数据转换为json存入消息队列,由于fastjson的速度快于jackson,这里替换为fastjson的一个实现 --> <bean id="jsonMessageConverter" class="com.jy.utils.FastJsonMessageConverter"></bean> <!-- 或者配置jackson --> <!-- <bean id="jsonMessageConverter" class="org.springframework.amqp.support.converter.Jackson2JsonMessageConverter" /> --><rabbit:template exchange="test-exchange" id="rabbitTemplate" connection-factory="connectionFactory" message-converter="jsonMessageConverter" /><!-- 消费者部分 --> <!-- 自定义接口类 --> <bean id="testHandler" class="com.rabbit.TestHandler"></bean><!-- 用于消息的监听的代理类MessageListenerAdapter --> <bean id="testQueueListenerAdapter" class="org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter" ><!-- 类名 --><constructor-arg ref="testHandler" /><!-- 方法名 --><property name="defaultListenerMethod" value="handlerTest"></property><property name="messageConverter" ref="jsonMessageConverter"></property> </bean><!-- 配置监听acknowledeg="manual"设置手动应答,它能够保证即使在一个worker处理消息的时候用CTRL+C来杀掉这个worker,或者一个consumer挂了(channel关闭了、connection关闭了或者TCP连接断了),也不会丢失消息。因为RabbitMQ知道没发送ack确认消息导致这个消息没有被完全处理,将会对这条消息做re-queue处理。如果此时有另一个consumer连接,消息会被重新发送至另一个consumer会一直重发,直到消息处理成功,监听容器acknowledge="auto" concurrency="30"设置发送次数,最多发送30次 --> <rabbit:listener-container connection-factory="connectionFactory" acknowledge="auto" concurrency="20"><rabbit:listener queues="spittle.alert.queue.1" ref="testQueueListenerAdapter" /><rabbit:listener queues="spittle.alert.queue.2" ref="testQueueListenerAdapter" /><rabbit:listener queues="spittle.alert.queue.2" ref="testQueueListenerAdapter" /> </rabbit:listener-container>
5,生产者(发送端)代码:
@Resource private RabbitTemplate rabbitTemplate; private String queueName; public void sendMessage(CommonMessage msg){try { logger.error("发送信息开始");System.out.println(rabbitTemplate.getConnectionFactory().getHost()); //发送信息 queueName交换机,就是上面的routingKey msg.getSource() 为 test_key rabbitTemplate.convertAndSend(queueName,msg.getSource(), msg);//如果是普通字符串消息需要先序列化,再发送消息//rabbitTemplate.convertAndSend(queueName,msg.getSource(), SerializationUtils.serialize(msg));logger.error("发送信息结束");} catch (Exception e) { e.printStackTrace();}}public void setQueueName(String queueName) {this.queueName = queueName; }
6,消费端代码:TestHandler 类
public class TestHandler {@Overridepublic void handlerTest(CommonMessage commonMessage) {System.out.println("DetailQueueConsumer: " + new String(message.getBody()));} }
其他exchangeType介绍:
fanOut:
<!-- Fanout 扇出,顾名思义,就是像风扇吹面粉一样,吹得到处都是。如果使用fanout类型的exchange,那么routing key就不重要了。因为凡是绑定到这个exchange的queue,都会受到消息。 --> <rabbit:fanout-exchange name="delayed_message_exchange" durable="true" auto-delete="false" id="delayed_message_exchange"> <rabbit:bindings> <rabbit:binding queue="test_delay_queue"/> </rabbit:bindings> </rabbit:fanout-exchange>
topic:如果说direct是将消息放到exchange绑定的一个queue里(一对一);fanout是将消息放到exchange绑定的所有queue里(一对所有);那么topic类型的exchange就可以实现(一对部分),应用场景就是打印不同级别的错误日志,我们的系统出错后会根据不同的错误级别生成error_levelX.log日志,我们在后台首先要把所有的error保存在一个总的queue(绑定了一个*.error的路由键)里,然后再按level分别存放在不同的queue。
<!-- 发送端不是按固定的routing key发送消息,而是按字符串“匹配”发送,接收端同样如此 --> <rabbit:topic-exchange name="message-exchange" durable="true" auto-delete="false" id="message-exchange"><rabbit:bindings><rabbit:binding queue="Q1" pattern="error.*.log" /><rabbit:binding queue="Q2" pattern="error.level1.log" /><rabbit:binding queue="Q3" pattern="error.level2.log" /></rabbit:bindings> </rabbit:topic-exchange>
routing key绑定如下图:
本文转自:
https://blog.csdn.net/nandao158/article/details/81065892
https://www.cnblogs.com/LipeiNet/p/6079427.html
https://www.toutiao.com/a6598154241037042189/?tt_from=mobile_qq&utm_campaign=client_share×tamp=1536277584&app=news_article&utm_source=mobile_qq&iid=43157585039&utm_medium=toutiao_android&group_id=6598154241037042189