【RocketMQ】消息可靠性保证

红太狼 2022-12-22 08:50 175阅读 0赞

在实际使用RocketMQ的时候我们并不能保证每次发送的消息都刚好能被消费者一次性正常消费成功， 可能会存在需要多次消费才能成功或者一直消费失败的情况，Broker该如何处理呢？

## 1.消息消费端的确认机制 ##

RocketMQ提供了ack机制（默认是手动ack），以保证消息能够被正常消费。为了保证消息肯定消费成功，只有使用方明确表示消费成功，RocketMQ才会认为消息消费成功，然后删除消息。中途断电，抛出异常等都不会认为成功

DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("my_consumer_group");
    consumer.registerMessageListener((MessageListenerConcurrently) (list, consumeOrderlyContext) -> {
        
        list.stream().forEach(messageExt -> System.out.println(new String(messageExt.getBody()))); 
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; // 签收，即手动ack
    });

所有消费者在设置监听的时候会提供一个回调，业务实现消费回调的时候

*  当回调方法中返回 ConsumeConcurrentlyStatus.CONSUME\_SUCCESS，RocketMQ才会认为这批消息（默认是1条）是消费完成的。
 *  如果这时候消息消费失败，例如数据库异常，余额不足扣款失败等一切业务认为消息需要重试的场景，只要返回ConsumeConcurrentlyStatus.RECONSUME\_LATER，RocketMQ就会认为这批消息消费失败了

## 2.消息的衰减重试 ##

为了保证消息肯定至少被消费一次，RocketMQ会把这批消息重新发回到broker，在延迟的某个时间点 （默认是10秒，业务可设置）后，再次投递到这个ConsumerGroup。而如果一直这样重复消费都持续失败到一定次数（默认16次），就会投递到DLQ死信队列。

![在这里插入图片描述][20201118192437115.png_pic_center]

应用可以监控死信队列来做人工干预可以修改broker-a.conf文件

messageDelayLevel = 1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h

## 3.重试消息的处理机制 ##

一般情况下我们在实际生产中是不需要重试16次，这样既浪费时间又浪费性能，而且可能会引起消息堆积（只有前一条确认被消费了，下一条消息才有机会被消费）。理论上当尝试重复次数达到我们想要的结果时如果还是消费失败，那么我们需要将对应消息进行记录，并且结束重复尝试

consumer.registerMessageListener((MessageListenerConcurrently) (list, consumeOrderlyContext) -> {
         
        for (MessageExt messageExt : list) {
          
            // 如果消息已经重发了3次
            if(messageExt.getReconsumeTimes()==3) {
             
                // 可以将对应的数据保存到数据库，后续再做处理
                System.out.println(messageExt.getMsgId()+","+messageExt.getBody());    
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;  // 签收，结束重试   
            }   
        }   
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; 
    });

[20201118192437115.png_pic_center]: /images/20221120/f287924a1bf44141a1ffe8369aec3c42.png