【消息中间件】02-RabbitMQ可靠性投递——消息落库、二次确认检测

可靠性投递

可靠性投递就是确保消息从生产者到消费者之间传输的可靠性，解决方案的思路可以从以下几个方面来着手：

确保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