RocketMQ消息重复消费的问题

川长思鸟来 2022-06-17 04:43 661阅读 0赞

重复消费的问题的一个可能的问题：消费者消费消息时产生了异常，并没有返回CONSUME\_SUCCESS标志。  
我急于寻找解决方法，结果百度的结果都是一期多个消费者问题云云，根本没有解决我的问题。  
我发现重复消费的消息和第一次消费的消息不同，多了一些重复消费的信息：  
reconsumeTimes=1，2，…10  
REAL\_TOPIC也会是：%RETRY%XXXXX  
这就是因为消息处理异常导致的消息重新消费，无路时重启服务端，还是通过mqadmin删除都没用，RocketMQ可以很好的保持消息，一定要消费成功才可以！  
官方对comsumerMessage方法的实现建议是：

> It is not recommend to throw exception,rather than returning ConsumeConcurrentlyStatus.RECONSUME\_LATER if consumption failure

无论如何，都不要抛出异常，如果需要重新消费，可以返回RECONSUME\_LATER主动要求重新消费。  
下面是我的代码，我加入了catch Exception根异常来捕获业务处理的异常。

consumer.registerMessageListener(new MessageListenerConcurrently() {
                    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
                        ConsumeConcurrentlyContext context) {
                        logger.debug(Thread.currentThread().getName() + " Receive New Messages: " + msgs + "%n");
                        MessagePack msgpack = new MessagePack();
                        for (MessageExt msg : msgs){
                            byte[] data = msg.getBody();
                            try {
                                RTMsgPack rtmsg = msgpack.read(data, RTMsgPack.class);
                                logger.debug("Receive a message:" + rtmsg);
                                anlysisRTMsgPack(rtmsg, engine);
                            } catch (IOException e) {
                                logger.error("Unpack RTMsg:", e);
                            } catch (Exception e1){
                                logger.warn("Unexcepted exception.", e1);
                            }
                        }
                        logger.debug("RETURN CONSUME SUCCESS.");
                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                    }
                });