🏆【Alibaba中间件技术系列】「RocketMQ技术专题」帮你梳理RocketMQ相关的消费问题以及原理分析总结

消息重复消费的问题

消息重复消费是各个MQ都会发生的常见问题之一,在一些比较敏感的场景下,重复消费会造成比较严重的后果,比如重复扣款等。

消息重复消费场景及解决办法

在什么情况下会发生RocketMQ的消息重复消费呢?

生产者重复发送场景

当系统的调用链路比较长的时候,比如,系统A调用系统B,系统B再把消息发送到RocketMQ中,在系统A调用系统B的时候。

如果系统B处理成功,但是迟迟没有将调用成功的结果返回给系统A的时候,系统A就会尝试重新发起请求给系统B,造成系统B重复处理,发起多条消息给RocketMQ造成重复消费。

消费者重复发送场景

在系统B发送消息给RocketMQ的时候,也有可能会发生和上面一样的问题,消息发送超时,结果系统B重试,导致RocketMQ接收到了重复的消息。

消费者重复发送场景

当RocketMQ成功接收到消息,并将消息交给消费者处理,如果消费者消费完成后还没来得及提交offset给RocketMQ,自己宕机或者重启了,那么RocketMQ没有接收到offset,就会认为消费失败了,会重发消息给消费者再次消费。

消费者没有立刻返回成功

重复消费的问题的一个可能的问题:消费者消费消息时产生了异常,并没有返回CONSUME_SUCCESS标志。

因为消息处理异常导致的消息重新消费,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;
                }
 });

设置CONSUME_FROM_LAST_OFFSET的问题

Consumer在消费时,会设置从哪里开始消费。默认是CONSUME_FROM_LAST_OFFSET,设置的值如代码所示。

public enum ConsumeFromWhere {
    /**
     * 一个新的订阅组第一次启动从队列的最后位置开始消费<br>
     * 后续再启动接着上次消费的进度开始消费
     */
    CONSUME_FROM_LAST_OFFSET,
    @Deprecated
   CONSUME_FROM_LAST_OFFSET_AND_FROM_MIN_WHEN_BOOT_FIRST,
    @Deprecated
    CONSUME_FROM_MIN_OFFSET,
    @Deprecated
    CONSUME_FROM_MAX_OFFSET,
    /**
     * 一个新的订阅组第一次启动从队列的最前位置开始消费<br>
     * 后续再启动接着上次消费的进度开始消费
     */
    CONSUME_FROM_FIRST_OFFSET,
    /**
     * 一个新的订阅组第一次启动从指定时间点开始消费<br>
     * 后续再启动接着上次消费的进度开始消费<br>
     * 时间点设置参见DefaultMQPushConsumer.consumeTimestamp参数
     */
    CONSUME_FROM_TIMESTAMP,
}
  • CONSUME_FROM_LAST_OFFSET:从最后的偏移量开始消费,是从该消费者上次消费到的位置开始消费。

    • 如果是一个新的消费者,就要根据这个client所属的消费组的情况来判断。

    • 如果所属的消费者组是新上线的,订阅的消息,最早的消息都没有过,RocketMQ的设计者认为,你这是一个新上线的业务,会强制从第一条消息开始消费。

    • 如果订阅的消息,已经产生了过期消息,那么才会从我们这个client启动的时间点开始消费。

ConsumeFromWhere这个参数只对一个新的消费者第一次启动时有效

  • CONSUME_FROM_FIRST_OFFSET:从最小偏移量开始消费,

  • CONSUME_FROM_TIMESTAMP:从某个时间开始消费。

  • 而判断是不是一个新的ConsumerGroup是在broker端判断。

  • 消费到哪个offset最先是存在Consumer本地的,定时和broker同步自己的消费offset。

  • broker在判断是不是一个新的consumergroup,就是查broker端有没有这个consumergroup的offset记录。

偏移量无效化

对于一个新的queue,这个参数也是没用的,都是从0开始消费。

所以,这就有了一个问题我已经设置了CONSUME_FROM_LAST_OFFSET,为什么还是重复消费了,可能你这不是新的consumergroup,也可能是个新的Queue。

重试队列和死信队列

  • 消费端,一直不回传消费的结果。RocketMQ认为消息没收到,consumer下一次拉取,broker依然会发送该消息。

  • 任何异常都要捕获返回:ConsumeConcurrentlyStatus.RECONSUME_LATER

RocketMQ会放到重试队列,TOPIC是:%RETRY%+COnsumerGroup的名字

  • 重试的消息在延迟的某个时间点(默认是10秒,业务可设置)后,再次投递到这个ConsumerGroup。

  • 而如果一直这样重复消费都持续失败到一定次数(默认16次),就会投递到DLQ死信队列,此时需要人工干预了。

/**
Batch consumption size
*/

private int consumeMessageBatchMaxSize = 1;

/**

Batch pull size
*/

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之后

生产环境不要改

messageDelayLevel = 5s 5s 5s 5s 5s 5s 5s 5s 5s 5s 5s 5s 5s 5s 5s 5s 5s 5s

16次之后,多了一个topic名为:%DLQ%+consumergroup

image

这个默认的16次,可以改,但是使用DefaultMQPullConsumer才可以修改。

DefaultMQPushConsumer不能修改此值。

consumeMessageBatchMaxSize 这个size是消费者注册的回调listener一次处理的消息数,默认是1,不是每次拉取的消息数(默认是32),这个不要搞混。

消息消费进度的更新

未来的文章会进行介绍相关进度更新的功能和分析

©著作权归作者所有,转载或内容合作请联系作者
  • 序言:七十年代末,一起剥皮案震惊了整个滨河市,随后出现的几起案子,更是在滨河造成了极大的恐慌,老刑警刘岩,带你破解...
    沈念sama阅读 199,902评论 5 468
  • 序言:滨河连续发生了三起死亡事件,死亡现场离奇诡异,居然都是意外死亡,警方通过查阅死者的电脑和手机,发现死者居然都...
    沈念sama阅读 84,037评论 2 377
  • 文/潘晓璐 我一进店门,熙熙楼的掌柜王于贵愁眉苦脸地迎上来,“玉大人,你说我怎么就摊上这事。” “怎么了?”我有些...
    开封第一讲书人阅读 146,978评论 0 332
  • 文/不坏的土叔 我叫张陵,是天一观的道长。 经常有香客问我,道长,这世上最难降的妖魔是什么? 我笑而不...
    开封第一讲书人阅读 53,867评论 1 272
  • 正文 为了忘掉前任,我火速办了婚礼,结果婚礼上,老公的妹妹穿的比我还像新娘。我一直安慰自己,他们只是感情好,可当我...
    茶点故事阅读 62,763评论 5 360
  • 文/花漫 我一把揭开白布。 她就那样静静地躺着,像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上,一...
    开封第一讲书人阅读 48,104评论 1 277
  • 那天,我揣着相机与录音,去河边找鬼。 笑死,一个胖子当着我的面吹牛,可吹牛的内容都是我干的。 我是一名探鬼主播,决...
    沈念sama阅读 37,565评论 3 390
  • 文/苍兰香墨 我猛地睁开眼,长吁一口气:“原来是场噩梦啊……” “哼!你这毒妇竟也来了?” 一声冷哼从身侧响起,我...
    开封第一讲书人阅读 36,236评论 0 254
  • 序言:老挝万荣一对情侣失踪,失踪者是张志新(化名)和其女友刘颖,没想到半个月后,有当地人在树林里发现了一具尸体,经...
    沈念sama阅读 40,379评论 1 294
  • 正文 独居荒郊野岭守林人离奇死亡,尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 35,313评论 2 317
  • 正文 我和宋清朗相恋三年,在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 37,363评论 1 329
  • 序言:一个原本活蹦乱跳的男人离奇死亡,死状恐怖,灵堂内的尸体忽然破棺而出,到底是诈尸还是另有隐情,我是刑警宁泽,带...
    沈念sama阅读 33,034评论 3 315
  • 正文 年R本政府宣布,位于F岛的核电站,受9级特大地震影响,放射性物质发生泄漏。R本人自食恶果不足惜,却给世界环境...
    茶点故事阅读 38,637评论 3 303
  • 文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹,春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 29,719评论 0 19
  • 文/苍兰香墨 我抬头看了看天上的太阳。三九已至,却和暖如春,着一层夹袄步出监牢的瞬间,已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 30,952评论 1 255
  • 我被黑心中介骗来泰国打工, 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留,地道东北人。 一个月前我还...
    沈念sama阅读 42,371评论 2 346
  • 正文 我出身青楼,却偏偏与公主长得像,于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子,可洞房花烛夜当晚...
    茶点故事阅读 41,948评论 2 341

推荐阅读更多精彩内容