原文:https://www.jianshu.com/p/1281f7fee69c测试
消费端,一直不回传 消费的结果。rocketmq认为消息没收到,consumer下一次拉取,broker依然会发送该消息。blog
因此,任何异常都要捕获返回ConsumeConcurrentlyStatus.RECONSUME_LATER队列
rocketmq会放到重试队列。io
这个重试TOPIC的名字是class
%RETRY%+consumergroup的名字im
在控制台上过一会就能够查到。异常
重试的消息在延迟的某个时间点(默认是10秒,业务可设置)后,再次投递到这个ConsumerGroup。而若是一直这样重复消费都持续失败到必定次数(默认16次),就会投递到DLQ死信队列,此时须要人工干预了。top
/**img
*/di
private int consumeMessageBatchMaxSize = 1;
/**
*/
private int pullBatchSize = 32;
consumeMessageBatchMaxSize 是批量消费的最大条数
pullBatchSize 是每次拉取的最大条数
在broker端的
private String messageDelayLevel = "1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h";
参数是设置重试的时间,即第一次1s以后,第二次5s以后
为了测试,改为5s,生产环境不要改
messageDelayLevel = 5s 5s 5s 5s 5s 5s 5s 5s 5s 5s 5s 5s 5s 5s 5s 5s 5s 5s
16次以后,多了一个topic
名为
%DLQ%+consumergroup
这个默认的16次,能够改。可是使用DefaultMQPullConsumer才能够修改。
DefaultMQPushConsumer不能修改此值。
顺便再说下,consumeMessageBatchMaxSize 这个size是消费者注册的回调listener一次处理的消息数,默认是1.不是每次拉取的消息数(默认是32),这个不要搞混。