MQ
- 概念
- 常见消息模型
- helloworld案例
- 实现
- 实现spring AMQP发送消息
- 实现spring AMQP接收消息
- 工作消息队列
- 实现
- 发布订阅模型
- Fanout Exchange
- 实现
- DirectExchange
- 实现
- TopicExchange
- 实现
- DirectExchange 和FanoutExchange的差异
- DirectExchange 和TopicExchange的差异
- 基于@RabbitListener注解声明队列有 哪些常用注
- 消息转换器
- 注意
- 同步调用
- 异步调用
- 安装
- SpringAMQP
- 特征
概念
MQ(MessageQueue):消息队列,事件驱动架构中的Broker
- channel:操作MQ的工具
- exchange:路由消息到队列
- queue:缓存消息
- virtual host: 虚拟主机,是对queue、exchange等资源逻辑分组
常见消息模型
helloworld案例
角色:
- publisher:消息发布者,将消息发送到队列queue
- queue:消息队列,负责接受并缓存消息
- consumer:订阅队列,处理队列中的消息
实现
实现spring AMQP发送消息
- 在父工程引入spring-amqp的依赖
<!--AMQP依赖,包含RabbitMQ--><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId></dependency>
-
在publisher服务中利用RabbitTemplate发送消息到simple.queue这个队列
- 在publisher服务中编写application.yml,添加mq连接信息
spring:rabbitmq:host: 127.0.0.1port: 5672virtual-host: /username: rootpassword: toor
- 在publisher服务中新建一个测试类,编写测试方法
@RunWith(SpringRunner.class) @SpringBootTest public class SpringAmqpTest {@Autowiredprivate RabbitTemplate rabbitTemplate;@Testpublic void testSimpleQueue(){String queueName = "simple.queue";String message = "hello,spring amqp!";rabbitTemplate.convertAndSend(queueName,message);} }
实现spring AMQP接收消息
- 在父工程引入spring-amqp的依赖
<!--AMQP依赖,包含RabbitMQ--><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId></dependency>
-
在consumer服务中编写消费逻辑,监听simple.queue。
- 在consumer服务中编写application.yml,添加mq连接信息
spring:rabbitmq:host: 127.0.0.1port: 5672virtual-host: /username: rootpassword: toor
- 在publisher服务中新建一个测试类,编写测试方法
@Component public class SpringRabbitListener {@RabbitListener(queues = "simple.queue")public void listenSimpleQueueMessage (String msg) throws InterruptedException{System.out.println("spring 消费者接收到消息:【"+msg+"】");} }
工作消息队列
作用: 提高消息处理速度,避免队列消息堆积。
实现
- 在publisher服务中定义测试方法,每秒产生50条消息,发送到simple.queue
@Autowiredprivate RabbitTemplate rabbitTemplate;@Testpublic void testWorkQueue() throws InterruptedException {String queueName = "simple.queue";String message = "hello,spring amqp!";for (int i = 0; i <= 50; i++) {rabbitTemplate.convertAndSend(queueName,message+i);Thread.sleep(20);}}
- 在consumer服务中定义两个消息监听者,都监听simple.queue队列
@RabbitListener(queues = "simple.queue")public void listenWorkQueueMessage1 (String msg) throws InterruptedException{System.out.println("spring 消费者1接收到消息:【"+msg+"】");Thread.sleep(20);}@RabbitListener(queues = "simple.queue")public void listenWorkQueueMessage2 (String msg) throws InterruptedException{System.err.println("spring 消费者1接收到消息:【"+msg+"】");//err输出为红色Thread.sleep(200);
- 消费者1每秒处理50条消息,消费者2每秒处理10条消息
- 修改application.yml文件,设置preFetch这个值,可以控制预取消息的上限
spring:rabbitmq:host: 127.0.0.1port: 5672virtual-host: /username: rootpassword: toorlistener:direct:prefetch: 1
发布订阅模型
概念: 与之前模型区别是,允许将同一消息发送给多个消费者。
实现方式: exchange(交换机)
exchange: 负责消息路由,不存储,路由失败则消息丢失
常见exchange类型:
- Fanout:广播
- Direct:路由
- Topic:话题
Fanout Exchange
Fanout Exchange:将接收到的消息路由到每一个跟其绑定的queue
实现
-
在consumer服务中,利用代码声明队列(Queue)、交换机(Exchange),并将两者绑定(Binding)
- SringAMQP提供了声明交换机、队列、绑定关系的API。
- 在consumer服务常见一个类,添加@configuration注解,并声明FanoutExchange、Queue和绑定关系对象Binding。
@Configuration public class FanoutConfig {//声明FanoutExchange交换机@Beanpublic FanoutExchange fanoutExchange(){return new FanoutExchange("root.fanout");}//声明第一个队列@Beanpublic Queue fanoutQueue1(){return new Queue("fanout.queue1");}//绑定队列1和交换机@Beanpublic Binding bindingQueue1(Queue fanoutQueue1,FanoutExchange fanoutExchange){return BindingBuilder.bind(fanoutQueue1).to(fanoutExchange);} //第二个同第一个 }
- SringAMQP提供了声明交换机、队列、绑定关系的API。
-
在consumer服务中,编写两个消费者方法,分别监听fanout.queue1和fanout.queue2
@RabbitListener(queues = "fanout.queue1")public void listenFanoutQueue1Message (String msg) throws InterruptedException{System.out.println("spring 消费者接收到消息:【"+msg+"】");}@RabbitListener(queues = "fanout.queue2")public void listenFanoutQueue2Message (String msg) throws InterruptedException{System.out.println("spring 消费者接收到消息:【"+msg+"】");}
- 在publisher中编写测试方法,向root.fanout发送消息
@Testpublic void testSendFanoutExchange(){String exchangeName = "root.fanout";String message = "hello,spring amqp!";rabbitTemplate.convertAndSend(exchangeName,message);}
DirectExchange
DirectExchange: 将接收到的消息更具规则路由到指定的Queue,因此称为路由模式(routes)
- 每一个Queue都与Exchange设置一个BindingKey
- 发布者发送消息时,指定消息的RoutingKey
- Exchange将消息路由到BindingKey与消息RoutingKey一致的队列
实现
- 利用@RabbitListener声明Exchange、Queue、RoutingKey
- 在consumer服务中编写两个消费者方法,分别监听direct.queue1和direct.queue2
@RabbitListener(bindings = @QueueBinding(value = @Queue(name="direct.queue1"),exchange = @Exchange(name = "root.direct",type = ExchangeTypes.DIRECT),key = {"red","yellow"}))public void listenDirectQueue1Message (String msg) throws InterruptedException{System.out.println("spring 消费者接收到direct.queue1的消息:【"+msg+"】");}@RabbitListener(bindings = @QueueBinding(value = @Queue(name="direct.queue2"),exchange = @Exchange(name = "root.direct",type = ExchangeTypes.DIRECT),key = {"red","yellow"}))public void listenDirectQueue2Message (String msg) throws InterruptedException{System.out.println("spring 消费者接收到direct.queue2的消息:【"+msg+"】");}
- 在publisher中编写测试方法,向root.direct发送消息
@Testpublic void testSendDirectExchange(){String exchangeName = "root.direct";String message = "hello,red!";rabbitTemplate.convertAndSend(exchangeName,"red",message);}
TopicExchange
TopicExchange: 与DirectExchange类似,区别在于routineKey必须是多个单词的列表,并且以== . ==分割。
Queue与Exchange指定BindingKey时可以使用通配符
# :代指0个或多个单词
*:代指一个单词
实现
- 利用@RabbitListener声明Exchange、Queue、RoutingKey
- 在consumer服务中编写两个消费者方法,分别监听topic.queue1和topic.queue2
@RabbitListener(bindings = @QueueBinding(value = @Queue(name="topic.queue1"),exchange = @Exchange(name = "root.topic",type = ExchangeTypes.TOPIC),key = "china.#"))public void listenTopicQueue1Message (String msg) throws InterruptedException{System.out.println("spring 消费者接收到topic.queue1的消息:【"+msg+"】");}@RabbitListener(bindings = @QueueBinding(value = @Queue(name="topic.queue2"),exchange = @Exchange(name = "root.topic",type = ExchangeTypes.TOPIC),key = "#.news"))public void listenTopicQueue2Message (String msg) throws InterruptedException{System.out.println("spring 消费者接收到topic.queue2的消息:【"+msg+"】");}
- 在publisher中编写测试方法,向root.topic发送消息
@Testpublic void testSendTopicExchange(){String exchangeName = "root.topic";String message = "hello,china.news!";rabbitTemplate.convertAndSend(exchangeName,"china.news",message);}
DirectExchange 和FanoutExchange的差异
- FanoutExchange将消息路由给每一个与之绑定的队列
- DirectExchange根据RoutingKey判断路由给哪个队列
- 如果多个队列具有相同的RoutingKey,则与Fanout功能类似
DirectExchange 和TopicExchange的差异
- TopicExchange的routineKey必须使用多个单词,以== . ==分割
- TopicExchange可以使用通配符
基于@RabbitListener注解声明队列有 哪些常用注
- @Queue
- @Exchange
消息转换器
设置JSON方式序列化:
发送消息
- 在publisher服务引入依赖
<dependency><groupId>com.fasterxml.jackson.dataformat</groupId><artifactId>jackson-dataformat-xml</artifactId><version>2.9.10</version>
</dependency>
- 在publisher服务声明MessageConverter
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;@Beanpublic MessageConverter jsonMessageConverter(){return new Jackson2JsonMessageConverter();}
接收消息
- 在consumer服务引入依赖
<dependency><groupId>com.fasterxml.jackson.dataformat</groupId><artifactId>jackson-dataformat-xml</artifactId><version>2.9.10</version>
</dependency>
- 在consumer服务声明MessageConverter
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;@Beanpublic MessageConverter jsonMessageConverter(){return new Jackson2JsonMessageConverter();}
- 定义消费者
@RabbitListener(queues = "object.queue")public void listenObjectQueueMessage(Map<String,Object> msg){System.err.println("spring 消费者1接收到消息:【"+msg+"】");//err输出为红色}
注意
MessageConverter默认是JDK序列化
接收方和发送方必须使用相同的MessageConverter
同步调用
优点: 时效性高
问题:
- 耦合度高
- 性能下降
- 资源浪费
- 级联失败
异步调用
实现方式:
- 事件驱动(常用)
优势:
- 服务解耦
- 性能提升,吞吐量提高
- 故障隔离。没有强依赖,不担心级联失败问题
- 流量削锋
缺点:
- 依赖Broker的可靠性、安全性、吞吐能力
- 架构复杂、业务没有明显的流程线,不好追踪管理
安装
docker pull rabbitmq:3-managementdocker run \-e RABBITMQ_DEFAULT_USER=root \-e RABBITMQ_DEFAULT_PASS=toor \--name mq \--hostname mq1 \-p 15672:15672 \-p 5672:5672 \-d \rabbitmq:3-management
SpringAMQP
AMQP(Advance Message Queuing Protocol):是用于在应用程序或之间传递业务消息的开放标准,该协议与语言平台无关,更符合微服务中独立性的要求
Spring AMQP: 是基于AMQP协议定义的一套API规范,提供了模板来发送和接收消息。包含两部分:
- spring-amqp:基础抽象
- spring-rabbit:底层默认实现
特征
- 监听器容器,用于异步处理入站消息
- 用于发送和接收消息的RabbitTemplate
- RabbitAdmin用于自动声明队列、交换和绑定