RocketMq 消息重试机制、死信队列( 三 ) _RocketMq

2.注册监听类
//最大重试次数consumer.setMaxReconsumeTimes(2);//顺序消费重试时间间隔consumer.setSuspendCurrentQueueTimeMillis(2000);
SuspendCurrentQueueTimeMillis表示重试的时间间隔，默认是1s，这里修改成2s
3.测试
@RunWith(SpringRunner.class)@SpringBootTest(classes = RocketmqApplication.class)class RocketmqApplicationTests {@Value("${rocketmq.consumer.topic}")private String topic;@Autowiredprivate RocketMQTemplate rocketMQTemplate;@Testpublic void testProducer(){String msg = "Orderly_test";rocketMQTemplate.convertAndSend(topic , msg);}}测试结果：

文章插图

可以看到三条结果，第一条是第一次消费的，其余两条是隔了2s重试的。重试2次之后这条数据就进入了死信队列。
顺序消费状态顺序消费目前也是两个状态：SUCCESS和
SUSPEND_CURRENT_QUEUE_A_MOMENT 。SUSPEND_CURRENT_QUEUE_A_MOMENT意思是先暂停消费一下，过SuspendCurrentQueueTimeMillis时间间隔后再重试一下，而不是放到重试队列里。

public enum ConsumeOrderlyStatus {/*** Success consumption*/SUCCESS,/*** Rollback consumption(only for binlog consumption)*/@DeprecatedROLLBACK,/*** Commit offset(only for binlog consumption)*/@DeprecatedCOMMIT,/*** Suspend current queue a moment*/SUSPEND_CURRENT_QUEUE_A_MOMENT;}

测试死信队列并发消费和顺序消费达到了最大重试次数之后就会放到死信队列。死信队列在一开始是不会被创建的，只有需要的时候才会被创建。就拿上面测试结果来看，进入到的死信队列就是%DLQ%consumer_group，进入死信队列的消息要收到处理。