RabbitMQ技能方案分析

方案分析

在上一篇文档中，详细陈诉了怎样通过Canal+MQ实现对分库分表的数据库和数据表举行数据同步，而在这个方案中，另有一个关键点是须要留意的：起首，数据增编削的信息是包管写入binlog的，Canal分析出增编削的信息后写入MQ，同步步调从MQ中读取消息，假如MQ中的消息丢失了数据将无法举行同步。
因此就须要对ＭＱ通报消息的可靠性举行研究
包管ＭＱ消息的可靠性分为两个方面：包管生产消息的可靠性、包管斲丧消息的可靠性。
包管生产消息可靠性

RabbitMQ提供生产者确认机制包管生产消息的可靠性，技能方案如下：
1. 起首发送消息的方法假如实行失败会举行重试，重试次数耗尽记载消息失败

zhilian-framework包中的zhilian-rabbitmq工程下面的client包的消息处置处罚类，用于发送消息

2. package com.zhilian.rabbitmq.client;
4. import cn.hutool.core.exceptions.ExceptionUtil;
5. import cn.hutool.core.util.CharsetUtil;
6. import cn.hutool.core.util.IdUtil;
7. import cn.hutool.core.util.StrUtil;
8. import com.zhilian.common.expcetions.MqException;
9. import com.zhilian.common.utils.DateUtils;
10. import com.zhilian.common.utils.JsonUtils;
11. import com.zhilian.common.utils.NumberUtils;
12. import com.zhilian.rabbitmq.dao.FailMsgDao;
13. import com.zhilian.rabbitmq.plugins.DelayMessagePostProcessor;
14. import com.zhilian.rabbitmq.plugins.RabbitMqListenableFutureCallback;
15. import lombok.extern.slf4j.Slf4j;
16. import org.springframework.amqp.core.Message;
17. import org.springframework.amqp.core.MessageBuilder;
18. import org.springframework.amqp.core.MessageDeliveryMode;
19. import org.springframework.amqp.rabbit.connection.CorrelationData;
20. import org.springframework.amqp.rabbit.core.RabbitTemplate;
21. import org.springframework.beans.factory.annotation.Autowired;
22. import org.springframework.retry.annotation.Backoff;
23. import org.springframework.retry.annotation.Recover;
24. import org.springframework.retry.annotation.Retryable;
25. import org.springframework.stereotype.Service;
27. import javax.annotation.Resource;
29. /**
30. * 消息处理类
31. *
32. */
33. @Slf4j
34. @Service
35. public class RabbitClient {
37.     @Resource
38.     private RabbitTemplate rabbitTemplate;
39.     @Autowired(required = false)
40.     private FailMsgDao failMsgDao;
42.     @Resource
43.     private RabbitClient rabbitClient;
45.     public void sendMsg(String exchange, String routingKey, Object msg) {
46.         rabbitClient.sendMsg(exchange, routingKey, msg, null, null, false);
47.     }
49.     /**
50.      * 发送消息 重试3次
51.      *
52.      * @param exchange   交换机
53.      * @param routingKey 路由key
54.      * @param msg        消息对象，会将对象序列化成json字符串发出
55.      * @param delay      延迟时间 秒
56.      * @param msgId      消息id
57.      * @param isFailMsg  是否是失败消息
58.      * @return 是否发送成功
59.      */
60.    
61. @Retryable(value = MqException.class, maxAttempts = 3, backoff = @Backoff(value = 3000, multiplier = 1.5), recover = "saveFailMag")
62.     public void sendMsg(String exchange, String routingKey, Object msg, Integer delay, Long msgId, boolean isFailMsg) {
63.         // 1.发送消息前准备
64.         // 1.1获取消息内容，如果非字符串将其序列化
65.         String jsonMsg = JsonUtils.toJsonStr(msg);
66.         // 1.2.全局唯一消息id，如果调用者设置了消息id，使用调用者消息id，如果为配置，默认雪花算法生成消息id
67.         msgId = NumberUtils.null2Default(msgId, IdUtil.getSnowflakeNextId());
68.         // 1.3.设置默认延迟时间，默认立即发送
69.         delay = NumberUtils.null2Default(delay, -1);
70.         log.debug("消息发送！exchange = {}, routingKey = {}, msg = {}, msgId = {}", exchange, routingKey, jsonMsg, msgId);
72.         // 1.4.构建回调
73.         RabbitMqListenableFutureCallback futureCallback = RabbitMqListenableFutureCallback.builder()
74.         .exchange(exchange)
75.         .routingKey(routingKey)
76.         .msg(jsonMsg)
77.         .msgId(msgId)
78.         .delay(delay)
79.         .isFailMsg(isFailMsg)
80.         .failMsgDao(failMsgDao)
81.         .build();
82.         // 1.5.CorrelationData设置
83.         CorrelationData correlationData = new CorrelationData(msgId.toString());
84.         correlationData.getFuture().addCallback(futureCallback);
86.         // 1.6.构造消息对象
87.         Message message = MessageBuilder.withBody(StrUtil.bytes(jsonMsg, CharsetUtil.CHARSET_UTF_8))
88.                 //持久化
89.                 .setDeliveryMode(MessageDeliveryMode.PERSISTENT)
90.                 //消息id
91.                 .setMessageId(msgId.toString())
92.                 .build();
94.         try {
95.             // 2.发送消息
96.             this.rabbitTemplate.convertAndSend(exchange, routingKey, message, new DelayMessagePostProcessor(delay), correlationData);
97.         } catch (Exception e) {
98.             log.error("send error:" + e);
99.             // 3.构建异常回调，并抛出异常
100.             MqException mqException = new MqException();
101.             mqException.setMsg(ExceptionUtil.getMessage(e));
102.             mqException.setMqId(msgId);
103.             throw mqException;
104.         }
105.     }
107.     /**
108.      * @param mqException mq异常消息
109.      * @param exchange    交换机
110.      * @param routingKey  路由key
111.      * @param msg         mq消息
112.      * @param delay       延迟消息
113.      * @param msgId       消息id
114.      */
115.     @Recover
116.     public void saveFailMag(MqException mqException, String exchange, String routingKey, Object msg, Integer delay, String msgId) {
117.         //发送消息失败，需要将消息持久化到数据库，通过任务调度的方式处理失败的消息
118.         failMsgDao.save(mqException.getMqId(), exchange, routingKey, JsonUtils.toJsonStr(msg), delay, DateUtils.getCurrentTime() + 10,  ExceptionUtil.getMessage(mqException));
119.     }
121. }

复制代码

@Retryable注解可实现方法实行失败举行重试，如下：

2. @Retryable(value = MqException.class, maxAttempts = 3, backoff = @Backoff(value = 3000, multiplier = 1.5), recover = "saveFailMag")

复制代码

属性分析如下：
value：抛出订定非常才会重试
include：和value一样，默以为空，当exclude也为空时，默认全部非常
exclude：指定不处置处罚的非常
maxAttempts：最大重试次数，默认3次
backoff：重试期待战略，默认利用@Backoff，@Backoff的value默以为1000L，我们设置为3000L；表现第一次失败后期待3秒后重试，multiplier（指定耽误倍数）默以为0，假如把multiplier设置为1.5表现每次期待重试时间是上一次的1.5倍，则第一次重试为3秒，第二次为4.5秒，第三次为6.75秒。
Recover: 设置回调方法名，当重试耗尽时，通过recover属性设置回调的方法名。通过@Recover注解界说重试失败后的处置处罚方法，在Recover方法中记载失败消息到数据库。
2. 通过MQ提供的生产者确认机制包管生产消息的可靠性

利用生产者确认机制须要给每个信息指定一个唯一ID，生产者确认机制通过异步回调的方式举行，包罗ConfirmCallback和Return回调。
ConfirmCallback：消息发送到Broker会有一个效果返回给发送者表现消息是否处置处罚乐成：
消息乐成投递到交换机，返回ack
消息未投递到交换机，返回nack
发送消息时指定回调对象

2. // 1.4.构建回调
3.         RabbitMqListenableFutureCallback futureCallback = RabbitMqListenableFutureCallback.builder()
4.         .exchange(exchange)
5.         .routingKey(routingKey)
6.         .msg(jsonMsg)
7.         .msgId(msgId)
8.         .delay(delay)
9.         .isFailMsg(isFailMsg)
10.         .failMsgDao(failMsgDao)
11.         .build();
12.         // 1.5.CorrelationData设置
13.         CorrelationData correlationData = new CorrelationData(msgId.toString());
14.         correlationData.getFuture().addCallback(futureCallback);
16.         // 1.6.构造消息对象
17.         Message message = MessageBuilder.withBody(StrUtil.bytes(jsonMsg, CharsetUtil.CHARSET_UTF_8))
18.                 //持久化
19.                 .setDeliveryMode(MessageDeliveryMode.PERSISTENT)
20.                 //消息id
21.                 .setMessageId(msgId.toString())
22.                 .build();

复制代码

回调类：RabbitMqListenableFutureCallback
假如没有返回ack则将消息记载到失败消息表，假如颠末重试后返回了ack分析消息发送乐成，此时将消息从失败消息表删除。

2. package com.zhilian.rabbitmq.plugins;
4. import cn.hutool.core.exceptions.ExceptionUtil;
5. import com.zhilian.common.utils.DateUtils;
6. import com.zhilian.rabbitmq.dao.FailMsgDao;
7. import lombok.Builder;
8. import org.springframework.amqp.rabbit.connection.CorrelationData;
9. import org.springframework.util.concurrent.ListenableFutureCallback;
11. @Builder
12. public class RabbitMqListenableFutureCallback implements ListenableFutureCallback<CorrelationData.Confirm> {
14.     //记录失败消息service
15.     private FailMsgDao failMsgDao;
17.     private String exchange;
18.     private String routingKey;
19.     private String msg;
20.     private Long msgId;
21.     private Integer delay;
23.     //是否是失败消息
24.     private boolean isFailMsg=false;
26.     @Override
27.     public void onFailure(Throwable ex) {
28.         if(failMsgDao == null) {
29.             return;
30.         }
31.         failMsgDao.save(msgId, exchange, routingKey, msg, delay, DateUtils.getCurrentTime() + 10, ExceptionUtil.getMessage(ex));
32.     }
34.     @Override
35.     public void onSuccess(CorrelationData.Confirm result) {
36.         if(failMsgDao == null){
37.             return;
38.         }
39.         if(!result.isAck()){
40.             // 执行失败保存失败信息，如果已经存在保存信息，如果不在信息信息
41.             failMsgDao.save(msgId, exchange, routingKey, msg, delay,DateUtils.getCurrentTime() + 10, "MQ回复nack");
42.         }else if(isFailMsg && msgId != null){
43.             // 如果发送的是失败消息，当收到ack需要从fail_msg删除该消息
44.             failMsgDao.removeById(msgId);
45.         }
46.     }
47. }

复制代码

Return回调：假如消息发送到交换机乐成了但是并没有到达队列，此时会调用ReturnCallback回调方法，在回调方法中我们可以收到失败的消息存入失败消息表以便举行补偿
要利用Return回调须要开启设置：（在shared-rabbitmq.yaml中设置rabbitMQ参数：

2. spring:
3.   rabbitmq:
4.     publisher-confirm-type: correlated
5.     publisher-returns: true
6.     template:
7.       mandatory: true

复制代码

分析：
publish-confirm-type：开启publisher-confirm，这里支持两种范例：
simple：同步期待confirm效果，直到超时
correlated：异步回调，界说ConfirmCallback，MQ返回效果时会回调这个ConfirmCallback
publish-returns：开启publish-return功能，同样是基于callback机制，不外是界说ReturnCallback
template.mandatory：界说消息路由失败时的战略。true，则调用ReturnCallback；false：则直接扬弃消息
对于发送消息失败之后将消息写入失败消息表的逻辑参考如下：

2. @Override
3. public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
4.     // 获取RabbitTemplate
5.     RabbitTemplate rabbitTemplate = applicationContext.getBean(RabbitTemplate.class);
6.     //定义returnCallback回调方法
7.     rabbitTemplate.setReturnsCallback(
8.         new RabbitTemplate.ReturnsCallback() {
9.             @Override
10.             public void returnedMessage(ReturnedMessage returnedMessage) {
11.                 byte[] body = returnedMessage.getMessage().getBody();
12.                 //消息id
13.                 String messageId = returnedMessage.getMessage().getMessageProperties().getMessageId();
14.                 String content = new String(body, Charset.defaultCharset());
15.                 log.info("消息发送失败，应答码{}，原因{}，交换机{}，路由键{},消息id{},消息内容{}",
16.                          returnedMessage.getReplyCode(),
17.                          returnedMessage.getReplyText(),
18.                          returnedMessage.getExchange(),
19.                          returnedMessage.getRoutingKey(),
20.                          messageId,
21.                          content);
22.                 if (failMsgDao != null) {
23.                     failMsgDao.save(NumberUtils.parseLong(messageId), returnedMessage.getExchange(), returnedMessage.getRoutingKey(), content, null, DateUtils.getCurrentTime(), "returnCallback");
24.                 }
25.             }
26.         }
27.     );
28. }

复制代码

包管斲丧信息可靠性

包管斲丧消息可靠性方案起首包管发送消息设置为恒久化，其次通过MQ的斲丧确认机制包管斲丧者斲丧乐成消息后再将消息删除。
起首设置消息恒久化，包管消息发送到MQ消息不丢失。详细须要设置交换机和队列支持恒久化，发送消息设置deliveryMode=2。
RabbitMQ是通过斲丧者回执来确认斲丧者是否乐成处置处罚消息的：斲丧者获取消息后，向RabbitMQ发送ACK回执，表明自己已经处置处罚完成消息，RabbitMQ收到ACK后删除消息。

斲丧消息失败重试3次，仍失败则将斲丧失败的消息放入失败消息数据库
通过使命调治扫描失败消息队列（错误消息队列）重新发送，到达肯定的次数还未乐成则由人工处置处罚
焦点代码实现：（从预界说的错误队列中取出之前处置处罚失败的消息，重新发送到消息本来的目的地点

2. package com.zhilian.rabbitmq.plugins;
4. import com.zhilian.common.utils.IoUtils;
5. import com.zhilian.common.utils.JsonUtils;
6. import com.zhilian.rabbitmq.domain.ErrorRabbitMqMessage;
7. import com.zhilian.rabbitmq.properties.RabbitmqProperties;
8. import com.rabbitmq.client.Channel;
9. import com.rabbitmq.client.GetResponse;
10. import lombok.extern.slf4j.Slf4j;
11. import org.springframework.amqp.core.Message;
12. import org.springframework.amqp.core.MessageBuilder;
13. import org.springframework.amqp.rabbit.core.RabbitTemplate;
15. import javax.annotation.PreDestroy;
16. import java.io.IOException;
17. import java.nio.charset.Charset;
19. @Slf4j
20. public class RabbitMqResender {
22.     private RabbitTemplate rabbitTemplate;
23.     private RabbitmqProperties rabbitmqProperties;
24.     private Channel channel;
26.     public RabbitMqResender(RabbitTemplate rabbitTemplate, RabbitmqProperties rabbitmqProperties) {
27.         this.rabbitTemplate = rabbitTemplate;
28.         this.rabbitmqProperties = rabbitmqProperties;
29.         channel = rabbitTemplate.getConnectionFactory()
30.         .createConnection()
31.         .createChannel(false);
33.     }
35.     /**
36.      * 从队列中获取一条数据并处理,如果没有消息，返回false，有消息返回true
37.      */
38.     public boolean getOneMessageAndProcess() {
40.         try {
41.             GetResponse response = channel.basicGet(rabbitmqProperties.getError().getQueue(), false);
42.             if(response == null) {
43.                 return false;
44.             }
45.             ErrorRabbitMqMessage errorRabbitMqMessage = JsonUtils.toBean(new String(response.getBody()), ErrorRabbitMqMessage.class);
46.             Message message = MessageBuilder.withBody(errorRabbitMqMessage.getMessage().getBytes(Charset.defaultCharset())).build();
47.             rabbitTemplate.send(errorRabbitMqMessage.getOriginExchange(), errorRabbitMqMessage.getOriginRoutingKey(), message);
48.             channel.basicAck(response.getEnvelope().getDeliveryTag(), false);
49.             return true;
50.         }catch (IOException e) {
51.             log.error("消息重发失败，e:",e);
52.             return false;
53.         }
54.     }
56.     @PreDestroy
57.     public void destory() {
58.         log.info("rabbitmq销毁...");
59.         IoUtils.close(channel);
61.     }
62. }

复制代码

2. 1. 核心功能
3. //错误消息的捞取
4. //从指定的错误队列（error.queue，由RabbitmqProperties配置）中拉取未被确认（unacknowledged）的消息。
6. 2. 消息解析与重发
7. //将错误队列中的消息体解析为ErrorRabbitMqMessage对象（包含原始交换机、路由键、消息内容等元数据）。
8. //使用RabbitTemplate将消息内容重新发送到原始的交换机（originExchange）和路由键（originRoutingKey）。
9. 3. 异常处理与可靠性
10. //解析失败：若消息无法解析为ErrorRabbitMqMessage，直接拒绝消息并丢弃（不重新入队），避免死循环。
11. //发送失败：若重发过程中抛出异常（如网络问题），拒绝消息并允许重新入队，等待下次重试。
12. //资源安全：通过@PreDestroy确保程序关闭时正确释放RabbitMQ的Channel和Connection，防止连接泄漏。

复制代码

RabbitMQ提供三个确认模式：
•manual：手动ack，须要在业务代码竣事后，调用api发送ack。
•auto：自动ack，由spring监测listener代码是否出现非常，没有非常则返回ack；抛出非常则返回nack
•none：关闭ack，MQ假定斲丧者获取消息后会乐成处置处罚，因此消息投递后立刻被删除
这三种确认模式各有各的特点：

• none模式下，消息投递是不可靠的，大概丢失
  
• auto模式雷同事件机制，出现非常时返回nack，消息回滚到mq；没有非常，返回ack
  
• manual：自己根据业务环境，判定什么时间该ack
  

本项目的设置：

2. spring:
3.     rabbitmq:
4.        ....
5.         listener:
6.             simple:
7.                 acknowledge-mode: auto #，出现异常时返回nack，消息回滚到mq；没有异常，返回ack
8.                 retry:
9.                     enabled: true # 开启消费者失败重试
10.                     initial-interval: 1000 # 初识的失败等待时长为1秒
11.                     multiplier: 10 # 失败的等待时长倍数，下次等待时长 = multiplier * last-interval
12.                     max-attempts: 3 # 最大重试次数
13.                     stateless: true # true无状态；false有状态。如果业务中包含事务，这里改为false

复制代码

本项目利用自动ack模式，当斲丧消息失败会重试，重试3次假如还失败会将消息投递到失败消息队列，由定时使命步调定时读取队列的消息,到达肯定的次数还未乐成则由人工处置处罚。
包管消息幂等性

斲丧者在斲丧消息时不免出现重复斲丧的环境，好比：斲丧者没有向MQ返回ack导致重复斲丧，以是斲丧者须要包管斲丧信息幂等性

幂等性是指岂论实行多少次其效果是划一的
好比：收到消息须要向数据新增一条记载，假如重复斲丧则会出现重复添加记载的题目。
根据场景分析办理方案：

• 查询操纵：自己具有幂等性
  
• 添加操纵：假如主键是自增则大概重复添加记载。
  

办理：包管幂等性可以设置数据库的唯一束缚，好比添加门生信息，将学号字段设置为唯一索引，纵然重复添加雷同门生，同一个学号只会添加一条记载

• 更新操纵：假如更新一个固定，好比update users set status = 1 where id = ？，自己就具有幂等性；
  

但是假如只允许更新乐成一次？
办理：可以利用token机制，发送消息前天生一个token写入redis，收到消息后分析出token，从redis查询token，假如乐成则分析没有斲丧，此时更新乐成，将token从redis删除，当重复斲丧雷同的消息时，由于token已经从redis删除不会再实行更新操纵

• 删除操纵：与更新操纵雷同，假如是删除某个详细的记载，好比：delete from users where id = ？，自己就具有幂等性，假如只允许删除乐成一次就可以接纳更新操纵雷同的方法（操纵缓存token机制）
  

根据以上分析：
为了包管消息幂等性，须要：

• 利用数据库的唯一束缚去控制
  
• 利用token机制：
  

• 
  
  • 消息具有唯一ID
    
  • 发送消息时将消息ID写入Redis
    
  • 斲丧时根据消息ID查询Redis判定是否已经斲丧，假如已经斲丧则不再斲丧
    
         
  

能否百分百包管MQ消息可靠性？

固然不能！
包管消息可靠性分两个方面：包管生产消息可靠性和包管斲丧消息可靠性
包管生产消息可靠性：
生产消息可靠性是通过MQ是否发送ack回执来举行判定的。假如发nack表现发送消息失败，此时会举行重发或记载到失败消息表，通过定时使命举行补偿发送。
假如Java步调并没有收到回执（如jvm历程非常竣事，或断电等因素），此时将无法包管生产消息的可靠性。
包管斲丧信息消息可靠性：
包管斲丧信息消息可靠性方案起首包管发送消息设置为恒久化，其次通过MQ的斲丧确认机制包管斲丧者斲丧消息乐成后再将消息删除。
固然设置了消息恒久化，消息进入MQ起首是在缓存存储，MQ会根据肯定的规则举行刷盘，（好比每隔几毫秒举行刷盘，假如在消息还没有生存到磁盘时MQ历程克制，此时将会丢失消息）固然可以利用镜像队列（用于在RabbitMQ集群中复制队列的消息，如许做的目的是进队伍列的可用性和容错性，以防止在单个节点故障时导致消息的丢失）但是也不能百分百包管消息不丢失。
我们固然加了很多的包管可靠性的机制，但是这只能去进步消息的可靠性，终极照旧不能做到百分百的可靠，因此利用MQ的场景就必须要思量消息可靠性题目的存在，做好补偿处置处罚使命
怎样包管Canal+MQ同步消息的序次性

起首明确Canal分析binlog日志信息按序次发到MQ的队列中，现在是要包管斲丧端怎样按序次斲丧队列中的消息。
生产中同一个服务会启动多个jvm历程，每个历程作为canal-mq-jzo2o-foundations的斲丧者,如下图：

现在对服务名称先修改为aa再修改为bb，在MQ中的有两个消息：
修改服务名称为aa
修改服务名称为bb
预期：终极将服务名称修改为bb
此时两条消息会被分发给两个jvm历程，假设“修改服务名称为aa”的消息发给jvm历程1，“修改服务名称为bb”的消息发给jvm历程2，两个历程分别去斲丧，此时无法控制两个消息的先后序次，大概导致服务名称终极并非修改为bb。
办理方法：
多个jvm历程监听同一个队列包管只有斲丧者活泼，即只有一个斲丧者吸收消息。
斲丧队列中的数据利用单线程。
队列须要增长x-single-active-consumer参数，表现否启用单一运动斲丧者模式。

设置完成查包管队列上存在SAC标识，如下图：

当有多个jvm历程都去监听该队列时，只有一个为活泼状态

假如利用x-single-active-consumer参数须要修改为如下代码：
在Queue中添加：arguments={@Argument(name="x-single-active-consumer", value = "true", type = "java.lang.Boolean") }
如下所示：

2. @RabbitListener(bindings = @QueueBinding(
3.     value = @Queue(name = "canal-mq-jzo2o-foundations",arguments={@Argument(name="x-single-active-consumer", value = "true", type = "java.lang.Boolean") }),
4.     exchange = @Exchange(name="exchange.canal-jzo2o",type = ExchangeTypes.TOPIC),
5.     key="canal-mq-jzo2o-foundations"),
6.                 concurrency="1"
7.                )
8. public void onMessage(Message message) throws Exception{
9.     parseMsg(message);
10. }

复制代码

concurrency=”1“表现 指定斲丧线程为1。