消息可靠性、堆积、重复、丢失解决方案

简要

在日常业务系统中，常常会使用消息作为通知，通过通知处理相关业务，例如：支付成功更新订单状态，当支付成功时，我们需要及时更新订单状态。为了业务解耦、应对高并发，引入消息中间件是很好的选择，接下来结合RabbitMq消息中间件，如何做到消息的可靠性、防止消息堆积、丢失。

RabbitMq通信模型

分为：发送者publisher、消费者consumer、消息服务器broker(交换机exchange、队列)
通信流程如下：

publisher发送消息给消息服务器
(1)消息服务器broker收到消息，缓存消息，通知publisher已收到
(2)由于网络原因消息服务器broker未收到消息
(1)通过交换机exchange将消息投递给队列queue，投递成功
(2)由于服务器宕机未能将消息投递给队列queue
消息者consumer成功收到来自队列queue推送的消息
(1)消息者consumer成功收到来自队列queue推送的消息，正常消费
(2)消息者consumer成功收到来自队列queue推送的消息，消费异常

可靠性解决方案

发送可靠性

保证消息不丢失，可靠抵达，可以使用事务消息，性能下降250倍，为此引入ACK确认机制

confirmCallback模式与returnCallback可以解决消息发送可靠性问题。

confirmCallback模式：当broker未收到来自publisher推送的消息时，会触发确认回调，此时ack=false，可在回调处理推送失败的消息。
returnCallback模式：当exchange未将消息成功推送至queue时，表示消息投递失败，会触发退回回调，可在回调处处理投递失败的消息。

// 确认回调模式
rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
    /**
     * 1. 只要消息抵达Broker就ack=true 或者消费者确认消息就ack=true
     * @param correlationData 当前消息的唯一关联数据（这个是消息的唯一id）
     * @param ack 消息是否成功收到
     * @param s 失败的原因
     */
    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String s) {
        /**
         * 1.做好消息确认机制（pulisher，consumer【手动ack】）
         * 2.每一个发送的消息都在数据库做好记录。定期将失败的消息再次发送一次
         */
        if (ack) {
            // 消息抵达Broker成功，更新消息的状态为已发送
            MqMessage smsMqMessage = new MqMessage();
            smsMqMessage.setModifiedTime(DateUtil.date());
            smsMqMessage.setMessageId(correlationData.getId());
            smsMqMessage.setSendNum(1);
            smsMqMessage.setMessageStatus(CommonConstant.MessageStatus.YFS);
            mqMessageMapper.updateByIdAndStatus(smsMqMessage, CommonConstant.MessageStatus.CWDD);
            return;
        }
        MqMessage smsMqMessage = new MqMessage();
        smsMqMessage.setMessageId(correlationData.getId());
        smsMqMessage.setModifiedTime(DateUtil.date());
        smsMqMessage.setMessageResult(s);
        smsMqMessage.setMessageStatus(CommonConstant.MessageStatus.CWDD);
        mqMessageMapper.updateById(smsMqMessage);
        // 服务收到了
        log.info("confirm...correlationData[" + correlationData + "]==>ack[" + ack + "]==>cause[" + s + "]");
    }

});

// 回退模式
rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {
    /**
     * 只要消息没有投递给指定的队列，就触发这个失败回调
     * @param message 投递失败的消息详细信息
     * @param replyCode 回复的状态码
     * @param replyText 回复的内容
     * @param exchange 当时这个消息发给哪个交换机
     * @param routingKey 当时这个消息用哪个路由键
     */
    @Override
    public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {

        if (ObjectUtil.isNotEmpty(message.getMessageProperties().getReceivedDelay())){
            return;
        }
        // 报错误了。修改数据库当前消息的状态->错误抵达
        String messageId = message.getMessageProperties().getHeader(PublisherCallbackChannel.RETURNED_MESSAGE_CORRELATION_KEY);
        MqMessage smsMqMessage = new MqMessage();
        smsMqMessage.setMessageId(messageId);
        smsMqMessage.setModifiedTime(DateUtil.date());
        smsMqMessage.setMessageResult(replyText);
        smsMqMessage.setMessageStatus(CommonConstant.MessageStatus.CWDD);
        mqMessageMapper.updateById(smsMqMessage);
        log.error("Fail...message[" + message + "]==>replyCode[" + replyCode + "]==>replyText[" + replyText + "]==>exchange[" + exchange + "]==>routingKey[" + routingKey + "]");

    }
});

消费可靠性

消费者获取到消息，成功处理，可以回复ACK给Broker

basic.ack用于肯定确认；broker将移除此消息
basic.nack用于否定确认；可以指定broker是否丢失此消息，可以批量
basic.reject用于否定确认；同上，但不能批量

默认，消息被消费者收到，就会从broker的queue中移除

queue无消费者，消息依然会被存储，直到消费者消费

消费者收到消息，默认会自动ack。但是如果无法确定此消息是否被处理完成，或者成功处理。我们可以开启手动ack模式

消息处理成功，ack()，接受下一个消息，此消息broker就会被移除
消息处理失败，nack()/reject()，重新发送给其他人进行处理，或者容错处理后ack
消息一直没有调用ack/nack方法，broker认为此消息正在被处理，不会投递给别人，此时客户端断开，消息不会被broker移除，会投递给别人

保证消费可靠性代码

// 消费异常启动重试机制（重试3次），重试无果持久化消息记录，保证消费回调异常消息不丢失
// 消息id
String messageId = message.getMessageProperties().getHeader(PublisherCallbackChannel.RETURNED_MESSAGE_CORRELATION_KEY);
// 由于异常未能执行del-key，防止计数器持久化，临时保留1分钟
Long increment = redisUtils.increment(RedisConstant.SMS_MQ_MESSAGE_KEY + messageId, RedisConstant.KEY_EXPIRE_TIME);
// 重试3次
if (increment >= 3){
    // 如果重试3次依然异常，保存消息记录至数据库，定时发送处理
    // 保证消息一定会发送出去。每一个都可以做好日志记录（给数据库保存每一个消息的详细信息），定期扫描数据库将失败的消息再发送一次
    MqMessage mqMessage = mqMessageMapper.selectById(messageId);
    if (ObjectUtil.isNotEmpty(mqMessage)){
        // 更新消息日志->推送状态、异常信息
        updateSmsMqMessageAfter(message, body, throwable);
    }else {
        // 保存消息日志，定时发放
        saveSmsMqMessageAfter(message, body, throwable);
    }

    // 拒绝接收，通知broker清理此消息
    channel.basicReject(message.getMessageProperties().getDeliveryTag(), false);
    // 删除key
    redisUtils.del(RedisConstant.SMS_MQ_MESSAGE_KEY + messageId);
}else {
    // 拒绝接收，重新入队
    channel.basicNack(message.getMessageProperties().getDeliveryTag(), false,true);
}
// 打印重试次数与错误摘要
log.error("receiver message fail，number of retries count=["+ increment +"] error info：{}" + throwable.getMessage());

堆积解决方案

场景：

消费者宕机积压
消费者消费能力不足积压
发送者发送流量太大

方案：

上线更多的消费者，进行正常消费
上线专门的队列消费服务，将消息先批量取出来，记录数据库，离线出来

消息重复解决方案

场景：

消费成功，事务已经提交，ack时，机器宕机。导致没有ack成功，Broker的消息重新由unack变为ready，并发送给其他消费者
消息消费成功，由于重试机制，自动又将消息发送出去
成功消费，ack时宕机，消息由unack变为ready，Broker又重新发送

方案：

消费者的业务消费接口应该设计为幂等性，比如扣库存有工作单的状态标志
使用防重表(redis/mysql)，发送消息每个都有业务的唯一标识，处理过就不用处理
rabbitMq的每一个消息都有redelivered字段，可以获取是否是被重新投递过来的，而不是第一次投递过来的

消息丢失解决方案

场景：

消费异常，消息被broker清理
未将消息发送至broker，消息丢失
未将消息投递值queue，消息丢失

方案：

发送消息时，先将消息进行备份至数据库

 public Response<?> sendMqMessage(MessageBodyReq req) {
        // 生成消息id
        String messageId = UUID.randomUUID().toString();
        // 1.记录消息日志，便于异常重试(保证消息一定会发送出去。每一个都可以做好日志记录（给数据库保存每一个消息的详细信息），定期扫描数据库将失败的消息再发送一次)
        MqMessage message = new MqMessage();
        message.setMessageId(messageId);
        message.setMessageContent(JSON.toJSONString(req.getBody()));
        message.setMessageStatus(CommonConstant.MessageStatus.XJ);
        message.setSendTime(DateUtil.date());
        message.setIsDelay(ObjectUtil.isNotEmpty(req.getDelayTime()) ? 1 : 2);
        message.setTimeValue(req.getDelayTime());
        message.setToExchange(req.getExchangeName());
        message.setRoutingKey(req.getRouteKey());
        message.setCreatedTime(DateUtil.date());
        message.setClassType(req.getBody().getClass().toString());
        mqMessageMapper.insert(message);
        try {
            MessageBody<Object> body = new MessageBody<>();
            body.setData(req.getBody());
            // 2.发送消息
            if (req.getDelayTime() != 0L) {
                // 发送延时消息
                rabbitTemplate.convertAndSend(req.getExchangeName(), req.getRouteKey(), body, sendMessage-> {
                        // 设置延时时间
                    sendMessage.getMessageProperties().setDelay((int) req.getDelayTime());
                        return sendMessage;
                },new CorrelationData(messageId));
            } else {
                // 发送消息
                rabbitTemplate.convertAndSend(req.getExchangeName(), req.getRouteKey(), body, new CorrelationData(messageId));
            }
        }catch (Exception e){
            // 网络等故障导致消息没有抵达broker，定期重发
            log.error("send message failure:{}",e.getMessage());

            return Response.error();
        }
        return Response.success();
    }