RocketMQ死信消息解决方案

当消息消费重试达到阈值后，将不再投递到消费者，而是存于名为"%DLQ%消费者组名"主题中为死信消息。

死信消息解决方案：

A、通过业务控制避免死信消息，当重试次数大于自定义时，将消息记录到特定位置或通知人工处理并完成消费，该方案使用较多。

@Test
void testConsumer() throws Exception {
    DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("retry-consumer-group");
    consumer.setNamesrvAddr(MQConstant.NAME_SRV_ADDR);
    consumer.subscribe("retryTopic", "*");
    consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
        MessageExt messageExt = msgs.get(0);
        System.out.println(new Date() + "----" + new String(messageExt.getBody()) + "----" + messageExt.getReconsumeTimes());
        try {
            // 业务处理
            int num = 10/0;
        } catch (Exception e) {
            // 控制重试，当重试次数大于自定义时，将消息记录到特定位置或通知人工处理并成功消费
            int reconsumeTimes = messageExt.getReconsumeTimes();
            if (reconsumeTimes < 3) {
                // 重试
                return ConsumeConcurrentlyStatus.RECONSUME_LATER;
            }
            // 不再重试
            System.out.println("将消息记录到特定位置或通知人工处理！");
        }
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    });
    consumer.start();
    System.in.read(); // 使用系统输入阻塞挂起JVM
}

B、消费死信主题消息，将消息记录到特定位置或通知人工处理并完成消费，但对于过多的死信主题则需要很多的消费者，该方案使用较少。

@Test
void testConsumerDead() throws Exception {
    DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("dead-consumer-group");
    consumer.setNamesrvAddr(MQConstant.NAME_SRV_ADDR);
    // 订阅死信消息
    consumer.subscribe("%DLQ%retry-consumer-group", "*");
    consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
        // 业务处理
        MessageExt messageExt = msgs.get(0);
        System.out.println(new Date() + "----" + new String(messageExt.getBody()) + "----" + messageExt.getReconsumeTimes());
        System.out.println("死信消息记录到特定位置或通知人工处理！");
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    });
    consumer.start();
    System.in.read(); // 使用系统输入阻塞挂起JVM
}