消息队列RabbitMQ

1. 简介与安装

RabbitMQ是基于Erlang语言开辟的开源消息通信中间件，支持AMQP，XMPP，SMTP，STOMP协议，消息延迟时微秒级别的。
Ubuntu系统RabbitMQ的安装

2. 基本概念

• Publisher 生产者，发送消息的一方
  
• Consumer 消耗者，接收消息的一方
  
• Queue 队列，存储消息
  
• Exchange 交换机，负责消息路由，生产者发送的消息由交换机负责投递到相应的队列。不具备存储消息的能力，因此如果没有任何队列与Exchange绑定，或者没有符合路由规则的队列，那么消息会丢失！
  
• VirtualHost 假造主机，起到数据隔离的作用，有各自的交换机和队列
  

3. SpringAMQP

• 导入Maven依靠
  
  1. <dependency>
  2.     <groupId>org.springframework.boot</groupId>
  3.     <artifactId>spring-boot-starter-amqp</artifactId>
  4. </dependency>

  复制代码
• SpringAMQP提供了RabbitTemplate工具，用于发送消息
  
• yml设置
  
  1. spring:
  2.   rabbitmq:
  3.     host: 127.0.0.1 # MQ部署的机器IP
  4.     port: 5672 # 端口
  5.     virtual-host: /test # 虚拟主机
  6.     username: admin # 用户名
  7.     password: admin # 密码

  复制代码
• RabbitMQ管理系统设置
  
  
  
  • 创建假造主机/test
    
  • 创建交换机test.direct
    
  • 创建队列test.queue
    
  • 将队列test.queue绑定到交换机test.direct
    
  
  
• 发送消息测试
  
  1. class LearnApplicationTests {
  3.     @Autowired
  4.     RabbitTemplate rabbitTemplate;
  5.    
  6.     @Test
  7.     void testSend() {
  8.         String exchange = "test.direct";
  9.         String msg = "hello RabbitMQ";
  10.         rabbitTemplate.convertAndSend(exchange, "", msg);
  11.     }
  12. }

  复制代码
• 接收消息测试
  
  1. import org.springframework.amqp.rabbit.annotation.RabbitListener;
  2. import org.springframework.stereotype.Component;
  4. /**
  5. * @author zyq
  6. */
  7. @Component
  8. public class SpringRabbitListener {
  9.     @RabbitListener(queues = "test.queue")
  10.     public void listenSimpleQueueMessage(String msg) {
  11.         System.out.println("消费者接收到消息：【" + msg + "】");
  12.     }
  13. }

  复制代码

4. 交换机范例

• Fanout交换机 广播交换机，将消息发送到全部绑定的队列
  
• Direct交换机 根据消息的Routing Key进行判定，只有队列的Routingkey与消息的 Routingkey完全一致，才会接收到消息
  
• Topic交换机 可以让队列在绑定BindingKey 的时候使用通配符
  
  
  
  • # 匹配一个或多个词
    
  • * 匹配一个词
    
  
  

5. 消息转换器

5.1 默认转换器

在数据传输时，发送的消息被序列化为字节发送给MQ，接收消息的时候，还会把字节反序列化为Java对象。 只不过，默认情况下Spring接纳的序列化方式是JDK序列化。众所周知，JDK序列化存在下列问题：

• 数据体积过大
  
• 有安全漏洞
  
• 可读性差
  

5.2 设置JSON转换器

• 引入Maven依靠
  1. <dependency>
  2.     <groupId>com.fasterxml.jackson.dataformat</groupId>
  3.     <artifactId>jackson-dataformat-xml</artifactId>
  4. </dependency>

  复制代码
• 设置Bean
  1. @Bean
  2. public MessageConverter messageConverter(){
  3.     return new Jackson2JsonMessageConverter();
  4. }

  复制代码

6 生产者的可靠性

一般情况下，只要生产者与MQ之间的网路连接顺畅，基本不会出现发送消息丢失的情况。少数情况下，大概出现投递的消息没有成功入队。
6.1 生产者超时重连机制

在生产者服务中进行如下设置

1. spring:
2.   rabbitmq:
3.     connection-timeout: 1s # 连接超时时间
4.     template:
5.       retry:
6.         enabled: true # 开启超时重连机制
7.         initial-interval: 1000ms # 初始等待时间
8.         multiplier: 1 # 等待时长倍数，下次等待时长 initial-interval * multiplier
9.         max-attempts: 3 # 重试次数

复制代码

当网络不稳定时，超时重连机制可以提高消息的发送成功率，但是SpringAMQP提供的重连机制时阻塞式的。不发起开启该功能，若业务必要，必要设置合理的等候时间和重试次数，也可以使用异步线程来实验发送消息的代码。
6.2 生产者确认机制

设置文件设置选项

1. spring:
2.   rabbitmq:
3.     publisher-confirm-type: correlated # 开启publisher confirm机制，并设置confirm类型
4.     # none：关闭confirm机制; simple：同步阻塞等待MQ的回执; correlated：MQ异步回调返回回执（推荐）
5.     publisher-returns: true # 开启publisher return机制

复制代码

• Publisher Return 消息成功到达交换机，但是路由失败时会触发ReturnCallback，往往时编程导致的，可以制止
  
  1. @Configuration
  2. public class MqConfig implements ApplicationContextAware {
  3.     @Override
  4.     public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
  5.         RabbitTemplate rabbitTemplate = applicationContext.getBean(RabbitTemplate.class);
  6.         rabbitTemplate.setReturnCallback((message, code, text, exchange, key) -> {
  7.             // 实现return callback
  8.             System.err.println("【Return Call】 message: " + message + ", replyText: " + text);
  9.         });
  10.     }
  11. }

  复制代码
• Publisher Confirm
  
  
  
  • 消息投递到交换机，但是路由失败，触发ReturnCallback，返回ACK
    
  • 临时消息（不必要持久化）投递到交换机并入队成功，返回ACK
    
  • 持久化消息投递到交换机，入队成功并完成持久化，返回ACK
    
  • 其他情况返回NACK，标识投递失败
    
  1. @Test
  2. void contextLoads() {
  3.     //  new CorrelationData(UUID.randomUUID().toString());
  4.     CorrelationData cd = new CorrelationData();
  5.     cd.getFuture().addCallback(new ListenableFutureCallback<CorrelationData.Confirm>() {
  6.         @Override
  7.         public void onFailure(Throwable throwable) {
  8.             // Future本身发生错误，一般不需要处理
  9.         }
  11.         @Override
  12.         public void onSuccess(CorrelationData.Confirm confirm) {
  13.             // Future处理成功
  14.             if (confirm.isAck()) {
  15.                 // 消息发送成功 ACK
  16.             } else {
  17.                 // 消息发送失败 NACK
  18.                 // 执行消息发送失败的业务逻辑
  19.             }
  20.         }
  21.     });
  22.     String exchange = "";
  23.     String routingKey = "";
  24.     String msg = "";
  25.     rabbitTemplate.convertAndSend(exchange, routingKey, msg, cd);
  26. }

  复制代码
  
  
• 总结
  生产者确认机制比力泯灭资源，一般不开启，不开启确认每秒钟可以投递数万的消息，而开启后只能投递数千。若业务必要高可靠性，只必要开启Publisher Confirm处理NACK的情况即可。
  

6. MQ的可靠性

消息到达MQ以后，如果MQ不能实时生存，也会导致消息丢失。
- MQ宕机；
- 内存空间不敷，引发MQ阻塞实验持久化；
6.1 数据持久化

• 交换机持久化（默认开启）；
  
• 队列持久化（默认开启）；
  
• 消息持久化（Delivery-mode必要指定为2，也就是持久化）
  - 若不开启消息持久化，在内存不敷时，会发生MQ阻塞写磁盘PageOut；
  - 若开启消息持久化，会同步将消息写到磁盘，MQ不会出现阻塞的征象，速率稍微慢一点点。
  

6.2 惰性队列 Lazy Queue

• 从3.6.0开始支持，从3.12开始默认使用该战略
  
• 接收到消息后直接写入磁盘（内存默认只保留2048条），消息消耗时才加载到内存，支持百万消息存储
  

7. 消耗者的可靠性

当RabbitMQ向消耗者投递消息以后，必要知道消耗者的处理状态如何。因为消耗者消耗消息大概出现故障，比如：

• 消息投递的过程中出现了网络故障
  
• 消耗者接收到消息后突然宕机
  
• 消耗者接收到消息后，因处理不当导致异常
  

7.1 消耗者确认机制

RabbitMQ提供了消耗者确认机制（Consumer Acknowledgement），当消耗者处理消息后，向RabbitMQ发送一个回执，告知RabbitMQ本身消息处理状态。回执有三种可选值：

• ack：成功处理消息，RabbitMQ从队列中删除该消息
  
• nack：消息处理失败，RabbitMQ必要再次投递消息
  
• reject：消息处理失败并拒绝该消息，RabbitMQ从队列中删除该消息
  

消息确认机制的实现方式

• none：不处理。即消息投递给消耗者后立刻ack，消息会立刻从MQ删除。非常不安全，不发起使用
  
• manual：手动模式。必要本身在业务代码中调用api，发送ack或reject，存在业务入侵，但更机动
  
• auto：自动模式。SpringAMQP利用AOP对我们的消息处理逻辑做了环绕加强，当业务正常实验时则自动返回ack. 当业务出现异常时，根据异常判定返回不同效果：
  
  
  
  • 如果是业务异常，会自动返回nack；
    
  • 如果是消息处理或校验异常，自动返回reject，比如发生MessageConversionException
    
  
  

7.2 失败重试机制

开启消耗者确认机制后，如果消息处理不停返回NACK，那么消息会反复进行入队和处理，会导致MQ压力飙升。
而开启失败重试机制后，消息会在当地重试，而不是重新入队，当地重试达到最大次数后，默认会返回reject丢弃消息。
在消耗者服务的设置文件中进行设置

1. spring:
2.   rabbitmq:
3.     listener:
4.       simple:
5.         retry:
6.           enabled: true # 开启消费者失败重试
7.           initial-interval: 1000ms # 初识的失败等待时长为1秒
8.           multiplier: 1 # 失败的等待时长倍数，下次等待时长 = multiplier * last-interval
9.           max-attempts: 3 # 最大重试次数
10.           stateless: true # true无状态；false有状态。如果业务中包含事务，这里改为false

复制代码

7.3 失败处理战略

当地重试达到最大次数后，默认会返回reject丢弃消息，而有些业务显然无法接受消息的丢失。MQ支持之界说重试次数耗尽后的处理战略

• RejectAndDontRequeueRecoverer：重试耗尽后，直接reject，丢弃消息，默认方式
  
• ImmediateRequeueMessageRecoverer：重试耗尽后，返回nack，消息重新入队
  
• RepublishMessageRecoverer：重试耗尽后，将失败消息投递到指定的交换机。（后续可进行人工处理）
  必要界说如下设置类
  

1. @Configuration
2. public class MqErrorConfig {
3.     private final static String ERROR_EXCHANGE = "error.direct";
4.     private final static String ERROR_QUEUE = "error.queue";
5.     private final static String ERROR_ROTING_KEY = "error";
7.     /**
8.      * 创建处理失败消息的交换机
9.      * @return
10.      */
11.     @Bean
12.     public DirectExchange errorExchange() {
13.         return new DirectExchange(ERROR_EXCHANGE);
14.     }
16.     /**
17.      * 创建存放失败消息的队列
18.      * @return
19.      */
20.     @Bean
21.     public Queue errorQueue() {
22.         return new Queue(ERROR_QUEUE);
23.     }
25.     /**
26.      * 交换机与队列绑定
27.      * @param errorQueue
28.      * @param errorExchange
29.      * @return
30.      */
31.     @Bean
32.     public Binding errorBinding(Queue errorQueue, DirectExchange errorExchange) {
33.         return BindingBuilder.bind(errorQueue).to(errorExchange).with(ERROR_ROTING_KEY);
34.     }
36.     /**
37.      * 注册处理失败消息处理策略
38.      * @param rabbitTemplate
39.      * @return
40.      */
41.     @Bean
42.     public MessageRecoverer messageRecoverer(RabbitTemplate rabbitTemplate) {
43.         return new RepublishMessageRecoverer(rabbitTemplate, ERROR_EXCHANGE, ERROR_ROTING_KEY);
44.     }
45. }

复制代码

7.4 业务幂等性方案

在步伐开辟中，则是指同一个业务，实验一次或多次对业务状态的影响是一致的。
7.4.1 唯一消息ID

• 每一条消息都生成一个唯一的id，与消息一起投递给消耗者。
  
• 消耗者接收到消息后处理本身的业务，业务处理成功后将消息ID生存到数据库
  
• 如果下次又收到相同消息，去数据库查询判定是否存在，存在则为重复消息放弃处理。
  进行如下设置，SpringAMQP会在消息头部自动添加唯一ID
  

1. @Bean
2. public MessageConverter messageConverter(){
3.     // 1.定义消息转换器
4.     Jackson2JsonMessageConverter jjmc = new Jackson2JsonMessageConverter();
5.     // 2.配置自动创建消息id，用于识别不同消息，也可以在业务中基于ID判断是否是重复消息
6.     jjmc.setCreateMessageIds(true);
7.     return jjmc;
8. }

复制代码

7.4.2 业务判定

非幂等性业务，会对数据进行更改，那么我们在实验业务逻辑前，可先判定数据记录是否处于未处理状态，比如可以根据订单的状态。
7.5 兜底战略

开启定时任务自动去查询数据库，判定数据有必要处理的数据。
8. 延迟消息

8.1 死信交换机

设计两个队列两个交换机，当消息过期时，消息会被投递到死信队列，只需监听死信队列即可。通过设置队列dead-letter-exchange指定过期的消息投递的交换机，也就是死信交换机。对于消息，通过expration指定过期时间。
然而，RabbitMQ的消息过期是基于追溯方式来实现的，也就是说当一个消息的TTL到期以后不肯定会被移除或投递到死信交换机，而是在消息恰恰处于队首时才会被处理。 当队列中消息堆积很多的时候，过期消息大概不会被按时处理，因此你设置的TTL时间不肯定正确。
8.2 DelayExchange插件

开启队列的delayed设置，而且在投递消息时设置delay时长。
延迟消息插件内部会维护一个当地数据库表，同时使用Elang Timers功能实现计时。如果消息的延迟时间设置较长，大概会导致堆积的延迟消息非常多，会带来较大的CPU开销，同时延迟消息的时间会存在误差。 因此，不发起设置延迟时间过长的延迟消息。
改进战略，将消息的delay时长分段，比如将延迟时间切割成10s 10s 10s 15s 15s …，大部分消息在前30s内就已经可以被消耗，不必要等到30分钟，可以有效防止消息堆积。
参考资料：https://www.bilibili.com/video/BV1mN4y1Z7t9/

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。