03-SpringAMQP入门
作者:互联网
三、SpringAMQP
SpringAMQP是基于RabbitMQ封装的一套模板,并且还利用SpringBoot对其实现了自动装配,使用起来非常方便
SpringAMQP的官方地址
AMQP
Spring AMQP
SpringAMQP提供了三个功能
- 自动声明队列、交换机及其绑定关系
- 基于注解的监听模式,异步接收消息
- 封装了RabbitTemplate工具,用于发送消息
3.1、Basic Queue 简单队列模型
在父工程引入依赖
<!--AMQP依赖,包含RabbitMQ--> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-amqp</artifactId> </dependency>
3.1.1、消息发送
-
①、配置MQ地址,在publisher服务的application.yml中添加配置
-
spring: rabbitmq: host: 192.168.222.135 # 主机名 port: 5672 # 端口号 virtual-host: /coolman # 虚拟主机 username: root # 用户名 password: root # 密码
-
-
②、在publisher服务中编写测试类,并利用RabbitTemplate实现消息发送
-
package cn.coolman.mq.springamqptest; import org.junit.jupiter.api.Test; import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.test.context.SpringBootTest; @SpringBootTest public class SimpleProducerTest { @Autowired private RabbitTemplate rabbitTemplate; @Test public void test01() { String queueName = "simpleQueue"; String message = "革命尚未完成,同志仍需努力"; rabbitTemplate.convertAndSend(queueName, message); System.out.println("发送简单消息:【" + message + "】完毕"); } }
-
3.1.2、消息接收
-
①、配置MQ地址,在consumer服务的application.yml中添加配置
-
spring: rabbitmq: host: 192.168.222.135 # 主机名 port: 5672 # 端口号 virtual-host: /coolman # 虚拟主机 username: root # 用户名 password: root # 密码
-
-
②、在consumer服务中定义队列
-
package cn.coolman.mq.config; import org.springframework.amqp.core.Queue; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; @Configuration public class RabbitQueueConfig{ /** * 定义队列,队列名称为:simpleQueue * @return */ @Bean public Queue simpleQueue(){ return new Queue("simpleQueue"); } }
-
-
③、在consumer服务中创建一个SpringRabbitListener类
-
package cn.coolman.mq.listener; import org.springframework.amqp.rabbit.annotation.RabbitListener; import org.springframework.stereotype.Component; @Component public class SimpleListener { /** * 方法的调用时机:如果队列中一旦出现了消息,马上就会调用这个方法去处理,并且会把这个消息传递给message参数 * @RabbitListener 指定监听的队列的名称 * @param message */ @RabbitListener(queues = "simpleQueue") public void listener1(String message) { System.out.println("监听器听到的消息:【"+ message + "】"); } }
-
3.1.3、测试
- 启动consumer服务,然后在publisher服务中运行测试代码,发送MQ消息
3.2、Work Queue 工作队列模型
Work Queues,也杯称为(Task Queues),任务模型
- 简单来说就是让多个消费者绑定到一个队列,共同消费队列中的消息
当消息处理比较耗时的时候,可能生产消息的速度会远远大于消费的速度。
长此以往,消息就会堆积越来越多,无法及时处理。
此时就可以使用work模型,多个消费者共同处理消息,速度就能大大提高
3.2.1、消息发送
-
这次我们循环发送,模拟大量消息堆积现象
-
在publisher服务中的SpringAmqpTest类中添加一个测试方法
-
@Test public void testWorkQueue() { String queueName = "workQueue"; String message = "good news"; for (int i = 1; i < 11; i++) { rabbitTemplate.convertAndSend(queueName, message + "\t" + i); System.out.println("发送简单消息:【" + message + "\t" + i + "】完毕"); } }
-
3.2.2、消息接收
-
修改RabbitQueueConfig类,将队列名称改为workQueue
-
package cn.coolman.mq.config; import org.springframework.amqp.core.Queue; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; @Configuration public class RabbitQueueConfig{ /** * 定义队列,队列名称为:simpleQueue * @return */ // @Bean // public Queue simpleQueue(){ // return new Queue("simpleQueue"); // } /** * 定义队列,队列名称为:workQueue * @return */ @Bean public Queue workQueue(){ return new Queue("workQueue"); } }
-
-
要模拟多个消费者绑定同一个队列,我们在consumer服务的SpringRabbitListener中添加1个新的方法,这两个方法绑定同一个队列
-
package cn.coolman.mq.listener; import org.springframework.amqp.rabbit.annotation.RabbitListener; import org.springframework.stereotype.Component; @Component public class SimpleListener { /** * 方法的调用时机:如果队列中一旦出现了消息,马上就会调用这个方法去处理,并且会把这个消息传递给message参数 * @RabbitListener 指定监听的队列的名称 * @param message */ @RabbitListener(queues = "workQueue") public void listener01(String message) { System.out.println("监听器01听到的消息:【"+ message + "】"); } @RabbitListener(queues = "workQueue") public void listener02(String message) throws InterruptedException { System.out.println("监听器02听到的消息:【"+ message + "】"); Thread.sleep(10000); } }
-
Listener02这个消费者sleep了10秒,模拟任务耗时
-
3.2.3、测试
- 启动ConsumerApplication后,执行publisher服务中刚编写的发送测试方法testWorkQueue
- 可以看到,我们发布的10条消息,居然平均分配给了两个消费者,要知道我们其中一个消费者可是每次都睡眠了10秒钟
- 这并没有考虑到消费者的处理能力,显然是有问题的
3.2.4、任务策略
能者多劳
-
在Spring中有个简单的配置,可以解决这个问题
-
修改consumer服务的application.yml文件,添加配置
-
spring: rabbitmq: listener: simple: prefetch: 1 # 每次只能获取一条消息,处理完成才能后去下一个消息
-
再进行测试
3.2.5、小结
- Work Queue模型的使用
- 多个消费者绑定到一个队列,同一条消息只会被一个消费者处理
- 通过设置prefetch来控制消费者预取的消息数量
3.3、Publish/Subscribe 发布/订阅模型
- 发布/订阅的模型如图所示
- 可以看到,在发布/订阅模型中,多了一个exchange角色,而且过程略有变化
- Publisher:生产者,也就是要发送消息的程序,但是不再发送到队列中,而是发给exchange(交换机)
- Exchange:交换机。一方面,接收生产者发送的消息;另一方面,知道如何处理消息,例如递交给某个特别队列、递交给所有队列、或者是将消息丢弃
- 到底如何操作,取决与Exchange的类型;Exchange有以下3中类型
- Fanout:广播,将消息交给所有绑定到交换机的队列
- Direct:定向,把消息交给符合指定routing key的队列
- Topic:通配符,把消息交给符合routing pattern(路由模式)的队列
- Consumer:消费者,与以前一样,订阅队列,没有变化
- Queue:消息队列也与以前一样,接收消息、缓存消息
- PS:
- Exchange(交换机)只负责转发消息,不具备存储消息的能力
- 因此如果没有任何队列与Exchange绑定,或者没有符合路由规则的队列,那么消息会丢失
3.3.1、Fanout 模型
- Fanout,中文意思是扇出,一般称为广播
- 在广播模式下,消息发送流程如下所示
- 1)可以有很多个队列
- 2)每个队列都要绑定到Exchange(交换机)
- 3)生产者发送的消息,只能发送到交换机,交换机来决定要发给哪个队列,生产者无法决定
- 4)交换机把消息发送给绑定过的所有队列
- 5)订阅队列的消费者都能拿到消息
①、声明队列和交换机
-
Spring提供了一个接口Exchange,来表示所有不同类型的交换机
-
在consumer中创建一个类,声明队列和交换机
-
package cn.coolman.mq.config; import org.springframework.amqp.core.Binding; import org.springframework.amqp.core.BindingBuilder; import org.springframework.amqp.core.FanoutExchange; import org.springframework.amqp.core.Queue; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; @Configuration public class RabbitQueueConfig{ // ========================定义发布订阅模式的交换机与队列,并且把队列绑定到交换机上================================= /** * 定义fanout的交换机 * @return */ @Bean // 这里的返回值对象已经存储到spring的容器中。 key 方法名: new FanoutExchange("fanoutExchange") public FanoutExchange fanoutExchange() { return new FanoutExchange("fanoutExchange"); } /** * 定义一个队列 */ @Bean public Queue fanoutQueue01() { return new Queue("fanoutQueue01"); } /** * 定义一个队列 */ @Bean public Queue fanoutQueue02() { return new Queue("fanoutQueue02"); } /** * 把队列绑定到交换机上 */ @Bean public Binding bindingQueueToExchange01(FanoutExchange fanoutExchange, Queue fanoutQueue01) { return BindingBuilder.bind(fanoutQueue01).to(fanoutExchange); } /** * 把队列绑定到交换机上 */ @Bean public Binding bindingQueueToExchange02(FanoutExchange fanoutExchange, Queue fanoutQueue02) { return BindingBuilder.bind(fanoutQueue02).to(fanoutExchange); } }
-
②、消息发送
-
在publisher服务的SpringAMQPTest类中添加测试方法
-
@Test public void testFanoutQueue() { String exchangeName = "fanoutExchange"; String message = "嘿嘿嘿嘿嘿嘿,fanout,i am coming "; rabbitTemplate.convertAndSend(exchangeName, "", message); System.out.println("发送成功!"); }
-
③、消息接收
-
在consumer服务的SpringRabbitListener中添加两个方法,作为消费者
-
package cn.coolman.mq.listener; import org.springframework.amqp.rabbit.annotation.RabbitListener; import org.springframework.stereotype.Component; @Component public class FanoutListener { /** * 方法的调用时机:如果队列中一旦出现了消息,马上就会调用这个方法去处理,并且会把这个消息传递给message参数 * @RabbitListener 指定监听的队列的名称 * @param message */ @RabbitListener(queues = "fanoutQueue01") public void listener01(String message) { System.out.println("监听器01听到的消息:【"+ message + "】"); } @RabbitListener(queues = "fanoutQueue02") public void listener02(String message) throws InterruptedException { System.out.println("监听器02听到的消息:【"+ message + "】"); } }
-
-
运行结果如下所示
- 由此可见,FanoutExchange会将消息无条件转发给每个绑定的队列
- 即routekey为空:
rabbitTemplate.convertAndSend(exchangeName, "", message);
④、小结
- 交换机的作用
- 接收publisher发送的消息
- 将消息按照规则路由到与之绑定的队列
- 不能缓存消息,路由失败,消息丢失
- FanoutExchange的会将消息路由到每个绑定的队列
- 声明队列、交换机、绑定关系的Bean是什么
- Queue
- FanoutExchange
- Binding
3.3.2、Direct 模型
- 在Fanout模式中,一条消息,会被所有订阅的队列消费。但是,在某些场景下,我们希望不同的消息被不同的队列消费。这时候就要用到Direct类型的Exchange
- 在Direct模型中
- 队列与交换机的绑定,不能是任意绑定了,而是要指定一个
RoutingKey
(路由Key)- Pulisher在向Excahnge发送消息的时候,也必须指定消息的
RoutingKey
- Exchange不再把消息交给每一个绑定的队列,而是根据消息的
RoutingKey
在进行判断,只有队列的RoutingKey
与消息的RoutingKey
完全一致,才会接收到消息
①、基于注解声明队列和交换机
-
基于@Bean的方式声明队列和交换机比较麻烦,Spring还提供了基于注解的方式来声明
-
在consumer的SpringRabbitListener中添加两个消费者,同时基于注解来声明队列和交换机
-
package cn.coolman.mq.listener; import org.springframework.amqp.core.ExchangeTypes; import org.springframework.amqp.rabbit.annotation.Exchange; import org.springframework.amqp.rabbit.annotation.Queue; import org.springframework.amqp.rabbit.annotation.QueueBinding; import org.springframework.amqp.rabbit.annotation.RabbitListener; import org.springframework.stereotype.Component; @Component public class DirectListener { @RabbitListener( bindings = @QueueBinding( exchange = @Exchange( name = "directExchange", type = ExchangeTypes.DIRECT ), value = @Queue("directQueue01"), key = "pig") ) public void listener01(String message) { System.out.println("监听器01听到的消息:【"+ message + "】"); } @RabbitListener( bindings = @QueueBinding( exchange = @Exchange( name = "directExchange", type = ExchangeTypes.DIRECT ), value = @Queue("directQueue02"), key = "dog") ) public void listener02(String message) { System.out.println("监听器02听到的消息:【"+ message + "】"); } }
-
②、消息发送
-
在publisher服务的SpringAMQPTest类中添加测试方法
-
@Test public void testDirectQueue() { String exchangeName = "directExchange"; String key1 = "pig"; String key2 = "dog"; String message = "I am a "; rabbitTemplate.convertAndSend(exchangeName, key1, message + key1); System.out.println("向directExchange交换机发送了路由key为【" + key1 + "】的消息【" + message + key1 + "】完毕!"); rabbitTemplate.convertAndSend(exchangeName, key2, message + key2); System.out.println("向directExchange交换机发送了路由key为【" + key2 + "】的消息【" + message + key2 + "】完毕!"); }
-
-
运行结果如下所示
- 可以明显看到,DirectExchange是根据
routekey
来分发消息的
- 可以明显看到,DirectExchange是根据
③、总结
- Direct交换机与Fanout交换机的差异
- Fanout交换机将消息路由给每一个与之绑定的队列
- Direct交换机根据RoutingKey判断路由给哪个队列
- 如果多个队列具有相同的RoutingKey,则与Fanout功能类似
- 基于@RabbitListener注解声明队列和交换机有那些常见的注解
- @Queue
- @Exchange
3.3.3、Topic 模型
- Topic模型的Exchange与Direct模型相比,都是可以根据
RoutingKey
把消息路由到不同的队列- 只不过Topic模型的Exchange可以让队列在绑定
RoutingKey
的时候使用通配符RoutingKey
一般是由一个或多个单词组成,多个单词之间以.
号分割,例如item.insert
- 通配符规则
#
:匹配一个或多个词*
:匹配不多不少,恰好一个词- 举例
item.#
:表示能够匹配item.spu.insert
或者item.spu
item.*
:只能匹配item.spu
- 再比如下图所示
- Queue1:绑定的是
china.#
,因此凡是以china.
开头的routing key
都会被匹配到。包括china.news
和china.weather
- Queue2:绑定的是
#.news
,因此凡是以.news
结尾的routing key
都会被匹配。包括china.news
和japan.news
- 案例需求
- ①、利用@RabbitListener声明Exchange、Queue、RoutingKey
- ②、在consumer服务中,编写两个消费者方法,分别监听
topic.queue1
和topic.queue2
- ③、在publisher中编写测试方法,向topicExchange发送消息
-
①、消息发送
-
@Test public void testTopicQueue() { String exchangeName = "topicExchange"; String routeKey1 = "coolman.hand"; String routeKey2 = "coolman.hand.hand.hand"; String message1 = "世纪之握!!!!"; String message2 = "究极无敌世纪之握!!!!"; rabbitTemplate.convertAndSend(exchangeName, routeKey1, message1); System.out.println("向【topicExchange】交换机发送 routeKey = 【coolman.hand】类型消息:【" + message1 + "】"); rabbitTemplate.convertAndSend(exchangeName, routeKey2, message2); System.out.println("向【topicExchange】交换机发送 routeKey = 【coolman.hand.hand.hand】类型消息:【" + message2 + "】"); }
②、消息接收
-
package cn.coolman.mq.listener; import org.springframework.amqp.core.ExchangeTypes; import org.springframework.amqp.rabbit.annotation.Exchange; import org.springframework.amqp.rabbit.annotation.Queue; import org.springframework.amqp.rabbit.annotation.QueueBinding; import org.springframework.amqp.rabbit.annotation.RabbitListener; import org.springframework.stereotype.Component; @Component public class TopicListener { @RabbitListener( bindings = @QueueBinding( exchange = @Exchange(name = "topicExchange", type = ExchangeTypes.TOPIC), value = @Queue("topic.queue1"), key = "coolman.*" ) ) public void listener01(String message) { System.out.println("【routeKey = " + "coolman.*】的监听器【topic.queue1】接收到消息:【" + message + "】"); } @RabbitListener( bindings = @QueueBinding( exchange = @Exchange(name = "topicExchange", type = ExchangeTypes.TOPIC), value = @Queue("topic.queue2"), key = "coolman.#" ) ) public void listener02(String message) { System.out.println("【routeKey = " + "coolman.#】的监听器【topic.queue2】接收到消息:【" + message + "】"); } }
-
运行结果如下所示
- 显然,
#
号可以匹配一个或多个词;*
号匹配不多不少,恰好一个词
- 显然,
③、小结
- Direct交换机和Topic交换机的差别
- Topic交换机接收的消息RoutingKey必须是多个单词,以
**.**
分割 - Topic交换机与队列绑定时的bingdingKey可以指定通配符
#
:代表0个或多个词*
:代表1个词
- Topic交换机接收的消息RoutingKey必须是多个单词,以
3.4、消息转换器
- 之前说过,Spring会把发送的消息序列化为字节发送给MQ,接收消息的时候,还会把字节反序列化为Java对象
- 只不过,默认情况下Spring采用的序列化方式是JDK序列化。众所周知,JDK序列化存在下列问题
- 数据体积过大
- 有安全漏洞
- 可读性差
- 接下来我们可以测试一下
3.4.1、测试默认转换器
-
我们修改消息发送的代码,发送一个Map对象
-
@Test public void testDefaultConvert() { String queueName = "simple.queue"; // 准备消息 HashMap<String, Object> message = new HashMap<>(); message.put("name", "coolman"); message.put("age", 8); message.put("power", "SSS+"); // 发送消息 rabbitTemplate.convertAndSend(queueName, message); }
-
-
停止consumer服务
-
发送消息后查看RabbitMQ控制台,查看map数据
-
显然,JDK序列化方式并不合适;我们希望消息体的体积更小、可读性更高,因此可以使用JSON方式来做序列化和反序列化
3.4.2、配置JSON转换器
-
在publisher和consumer两个服务中都引入依赖
-
<dependency> <groupId>com.fasterxml.jackson.dataformat</groupId> <artifactId>jackson-dataformat-xml</artifactId> <version>2.9.10</version> </dependency> <dependency> <groupId>com.fasterxml.jackson.core</groupId> <artifactId>jackson-databind</artifactId> <version>2.9.10.5</version> </dependency>
-
-
配置消息转换器,在启动类中添加一个Bean即可,如下所示
-
/** * 消息转换器 * @return */ @Bean // amqp的MessageConverter public MessageConverter jsonMessageConverter() { return new Jackson2JsonMessageConverter(); } }
-
-
再次向
simple.queue
发送消息,再在控制台查看,如下图所示 -
显然,使用JSON进行序列化,可以有效减小体积和提高可读性
标签:03,SpringAMQP,入门,队列,springframework,消息,import,org,message 来源: https://www.cnblogs.com/OnlyOnYourself-lzw/p/16491791.html