可靠性投递
可靠性投递就是确保消息从生产者到消费者之间传输的可靠性，解决方案的思路可以从以下几个方面来着手：
- 确保MQ收到生产者消息
- 确保消费者收到并处理完消息
- 完善的消息补偿机制
RabbitMQ自带的confirm机制可以实现前两点，我们需要自己实现消息补偿机制。
目前的消息
方案一：消息落库
消息落库重发是基于RabbitMQ的confirm机制，在消息发送失败后自动重发。该方案只能保证消息从生产者到MQ之间的可靠性投递
架构图：
消息落库.png
具体步骤：
Step 1：业务数据和消息数据分别入库，标记消息发送中、记录消息超时时间
Step 2：发送消息
Step3、4：异步监听MQ应答；响应成功，则标记消息发送成功；失败则，标记消息发送失败。
Step5、6、7：取出未发送成功(status:0)且响应超时的消息，重新发送。若发送次数大于3，则标记位失败消息，通知人工处理。
实例：
下面是一个发送订单消息的实例：创建订单同时将订单数据发送到消息中间件
只给出部分代码，需要详细代码请去GitHub下载
仓库地址：https://github.com/Tooi6/demo-rabbitmq-retry.git
-
消息日志实体
保存消息数据、标记消息状态。
public class BrokerMessageLog {
/**
* 消息uid
*/
private String messageId;
/**
* 消息内容（JSON保存）
*/
private String message;
/**
* 重试次数，阈值：3
*/
private Integer tryCount;
/**
* 消息状态，0：未发送成功、1：发送成功、2：失败消息
*/
private String status;
/**
* 超时时间（下次重发时间）
*/
private Date nextRetry;
/**
* 创建时间
*/
private Date createTime;
/**
* 更新时间
*/
private Date updateTime;
}
- 创建订单方法
@Service
public class OrderService {
@Autowired
private OrderMapper orderMapper;
@Autowired
private BrokerMessageLogMapper brokerMessageLogMapper;
@Autowired
private RabbitOrderSender rabbitOrderSender;
public void createOrder(Order order) throws Exception {
// 使用当前时间当做订单创建时间（为了模拟一下简化）
Date orderTime = new Date();
// 插入业务数据
orderMapper.insert(order);
// 插入消息记录表数据
BrokerMessageLog brokerMessageLog = new BrokerMessageLog();
// 消息唯一ID
brokerMessageLog.setMessageId(order.getMessageId());
// 保存消息整体 转为JSON 格式存储入库
brokerMessageLog.setMessage(FastJsonConvertUtil.convertObjectToJSON(order));
brokerMessageLog.setTryCount(0);
// 设置消息状态为0 表示发送中
brokerMessageLog.setStatus("0");
// 设置消息未确认超时时间窗口为 一分钟
brokerMessageLog.setNextRetry(DateUtils.addMinutes(orderTime, Constants.ORDER_TIMEOUT));
brokerMessageLog.setCreateTime(new Date());
brokerMessageLog.setUpdateTime(new Date());
brokerMessageLogMapper.insert(brokerMessageLog);
// 发送消息
rabbitOrderSender.sendOrder(order);
}
}
- 消息生产者：
发送消息到MQ，并处理Ack回调
@Component
public class RabbitOrderSender {
@Autowired
private RabbitTemplate rabbitTemplate;
@Autowired
private BrokerMessageLogMapper brokerMessageLogMapper;
/**
* ACK回调函数
*/
final ConfirmCallback confirmCallback = new RabbitTemplate.ConfirmCallback() {
@Override
public void confirm(CorrelationData correlationData, boolean ack, String cause) {
System.err.println("correlationData: " + correlationData);
String messageId = correlationData.getId();
if (ack) {
//返回成功 则进行更新消息日志 status=1
brokerMessageLogMapper.changeBrokerMessageLogStatus(messageId, Constants.ORDER_SEND_SUCCESS, new Date());
} else {
//失败则进行具体的后续操作:重试 或者补偿等手段
System.err.println("异常处理...");
}
}
};
/**
* 发送消息，设置回调函数
* @param order 业务实体
* @throws Exception
*/
public void sendOrder(Order order) throws Exception {
// 设置ACK回调
rabbitTemplate.setConfirmCallback(confirmCallback);
//消息唯一ID
CorrelationData correlationData = new CorrelationData(order.getMessageId());
rabbitTemplate.convertAndSend("order-exchange", "order.ABC", order, correlationData);
}
}
- 定时任务：
重发超时消息
@Component
public class RetryMessageTasker {
@Autowired
private RabbitOrderSender rabbitOrderSender;
@Autowired
private BrokerMessageLogMapper brokerMessageLogMapper;
@Scheduled(initialDelay = 5000, fixedDelay = 10000)
public void reSend() {
// 取出 status=0 且响应超时的消息
List<BrokerMessageLog> list = brokerMessageLogMapper.query4StatusAndTimeoutMessage();
list.forEach(messageLog -> {
if (messageLog.getTryCount() >= 3) {
// 重试3次，更新消息状态 status=2
brokerMessageLogMapper.changeBrokerMessageLogStatus(messageLog.getMessageId(), Constants.ORDER_SEND_FAILURE, new Date());
} else {
// 重新发送，tryCount+1
brokerMessageLogMapper.update4ReSend(messageLog.getMessageId(), new Date());
Order reSendOrder = FastJsonConvertUtil.convertJSONToObject(messageLog.getMessage(), Order.class);
try {
rabbitOrderSender.sendOrder(reSendOrder);
} catch (Exception e) {
e.printStackTrace();
System.err.println("-----------异常处理-----------");
}
}
});
}
}
方案一的缺点：
- 发送消息前需要2次DB操作，影响并发性能
- 只能保证消息从生产者到MQ的可靠性投递
方案二：二次确认检测
二次确认检测是基于延时投递机制实现的，该方案能够保证消息从生成者端到消费者的可靠性投递。
同时对比方案一，生产者操作数据库的次数相对较少，并发性能较高。
架构图：
延时投递.png
实例：
TODO
