RocketMQ重试队列与死信队列简介

Posted 2022-07-29 Hollis Chuang

1、消费者异常了怎么办？

假设我们使用RocketMQ作为消息中间件，传输订单相关的数据，消费者拿到数据后，执行一些后续处理，比如调用物流系统，准备发货。

如果这时候，物流系统的数据库宕机了，就必然会导致我们从MQ里获取到的消息没法进行处理。

图1 消费者异常

针对这样的异常场景我们应该怎么处理？物流系统应该怎么对消息进行重试？重试多少次才行？万一反复重试都没法成功，这个时候怎么办？消息能直接给扔了吗？

消费者系统一般获取消息，执行如下的代码：

consumer.registerMessageListener(
    new MessageListenerConcurrently() 
        @Override
        public ConsumeConcurrentlyStatus consumeMessage(
            List<MessageExt> msgs,
            ConsumeConcurrentlyContext context) 
                // 后续消息处理逻辑
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            
        
    
);

消费者中我们注册了一个监听器回调函数，当Consumer获取消息后，就会交给我们的回调函数来进行处理。如果处理完了，就返回一个ConsumeConcurrentlyStatus.CONSUME_SUCCESS，提交这批消息的offset到broker去，然后继续从broker获取下一批消息来进行处理。

如果上面代码回调函数中，对一批消息处理的时候，数据库宕机了，导致订单无法完成发货，我们还能返回CONSUME_SUCCESS吗？

如果你返回的话，下一次就会处理下一批消息，但是这批消息其实没处理成功，此时必然导致这批消息就丢失了。

2、消费者消息处理异常，可以返回RECONSUME_LATER

如果我们因为数据库宕机等问题，对这批消息没法完成处理，我们就应该返回一个RECONSUME_LATER状态。

他的意思是，我现在没法完成这批消息的处理，你过段时间再次给我这批消息让我重新试一下！

所以我们的代码应改成这样：

consumer.registerMessageListener(
    new MessageListenerConcurrently() 
        @Override
        public ConsumeConcurrentlyStatus consumeMessage(
            List<MessageExt> msgs,
            ConsumeConcurrentlyContext context) 
                try 
                    // 后续消息处理逻辑
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                 catch (Exception e) 
                    // 比如因为数据库宕机了，没法对消息完成处理
                    // 此时可以返回一个稍后重试的消费状态
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                
            
        
    
);

如果消息处理失败了，就返回RECONSUME_LATER状态，让RocketMQ稍后再重新把这批消息给我，让我重试对这批消息进行处理！

那么RocketMQ在收到你返回的RECONSUME_LATER状态之后，是如何让你进行消费重试的呢？

3、RocketMQ如何让你进行重试消费的？

简单来说，RocketMQ会有一个针对你这个ConsumerGroup的重试队列，如果你返回了RECONSUME_LATER状态，他就会把你这批消息放到你这个消费组的重试队列中去

比如你的消费组是"WMSConsumerGroup"，就是物流系统消费组，那么他就会有一个“%RETRY%WMSConsumerGroup”，这个名字的重试队列。

图2 RocketMQ重试队列

然后过一段时间，重试队列中的消息会再次给我们，让我们进行处理。如果再次失败，又返回了RECONSUME_LATER，那么会再过一段时间让我们再次进行处理，默认最多重试16次！每次重试之间的间隔时间是不一样的，这个间隔时间可以如下进行配置：

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

意思就是，第一次重试是1秒后，第二次重试是5秒后，第三次重试是10秒后，第四次重试是30秒后，第五次重试是1分钟后，以此类推，最多重试16次！

4、消费组重试一直失败怎么办？

那么如果在16次重试范围内消息处理成功了，自然就没问题了，但是如果你对一批消息重试了16次还是无法成功处理呢？

这个时候就需要另外一个队列了，叫做死信队列。

所谓的死信队列，顾名思义，就是死掉的消息就放这个队列里。

一批消息交给你处理，你重试了16次还一直没处理成功，就不要继续重试这批消息了，RocketMQ就认为他们死掉了。然后这批消息会自动进入死信队列。

死信队列的名字是“%DLQ%WMSConsumerGroup”，我们其实在RocketMQ的管理后台上都是可以看到的。

图3 死信队列

那么对死信队列中的消息我们应该怎么处理？

其实这个就看你具体的需求了，比如我们可以专门开一个后台线程，订阅“%DLQ%WMSConsumerGroup”这个死信队列，对死信队列中的消息进行不停的重试。

5、总结

这篇文章我们搞清楚了一个生产环境下可以说必须考虑的问题，就是消费者底层的一些依赖可能有故障了，比如数据库宕机，此时你没办法完成消息的处理了，那么可以通过返回一些状态让消息进入RocketMQ自带的重试队列，如果反复重试还是不行，可以让消息进入RocketMQ自带的死信队列，再针对死信队列中的消息进行单独的处理。

有道无术，术可成；有术无道，止于术

欢迎大家关注Java之道公众号

好文章，我在看❤️