0
  • 聊天消息
  • 系统消息
  • 评论与回复
登录后你可以
  • 下载海量资料
  • 学习在线课程
  • 观看技术视频
  • 写文章/发帖/加入社区
会员中心
创作中心

完善资料让更多小伙伴认识你,还能领取20积分哦,立即完善>

3天内不再提示

扒一扒RocketMQ中会导致消息重复消息的原因

jf_ro2CN3Fa 来源:三友的java日记 2023-05-06 12:25 次阅读

消息发送异常时重复发送

首先,我们来瞅瞅RocketMQ发送消息和消费消息的基本原理。

5704e1c2-eb16-11ed-90ce-dac502259ad0.png

如图,简单说一下上图中的概念:

Broker,就是RocketMQ的服务端,如上图就有两个服务实例

Topic就是一类消息集合的名字

Queue就是Topic的对应的队列,消息都存在Queue上,每个Topic都会有自己的几个Queue

所以,整个消息发送和消费过程大致如下:

生产者在发送消息之前根据负载均衡策略(默认是轮询)选择一个Queue,然后跟这个Queue所在的机器建立连接,把消息发送到这个Queue上

消费者只要消费这个Queue,那么就能消费到消息

在正常情况下,生产者的确是按照这个方式来发送消息的

但是当出现了异常时,这种异常包括消息发送超时、响应超时等等,RocketMQ为了保证消息成功发送,会进行消息发送的重试操作,默认情况下会最多会重试两次

571c1aa4-eb16-11ed-90ce-dac502259ad0.png

重试操作比较简单,就是选择另一台机器的Queue来发送。

虽然重试操作可以很大程度保证消息能够发送成功,但是同时也会带来消息重复发送的问题。

举个例子,假设生产者向A机器发送消息,发生了异常,响应超时了,但是就一定代表消息没发成功么?

不一定,有可能会出现服务端的确接受到并处理了消息,但是由于网络波动等等,导致生产者接收不到服务端响应的情况,此时消息处理成功了,但是生成者还是以为发生了异常

此时如果发生重试操作,那么势必会导致消息被发送了两次甚至更多次,导致服务端存了多条相同的消息,那么就一定会导致消费者重复消费消息

消费消息抛出异常

在RocketMQ的并发消费消息的模式下,需要用户实现MessageListenerConcurrently接口来处理消息

57c9e24c-eb16-11ed-90ce-dac502259ad0.png

当消费者获取到消息之后会调用MessageListenerConcurrently的实现,传入需要消费的消息集合 msgs,这里提到的msgs很重要

57eecf1c-eb16-11ed-90ce-dac502259ad0.png

如上代码,当消息消费出现异常的时候,status就会为null,后面就会将status设置成为RECONSUME_LATER。

RECONSUME_LATER翻译成功中文就是稍后重新消费的意思

所以从这可以看出,一旦抛出异常,那么消息之后就可以被重复消息。

到这其实可能有小伙伴觉得消息消费失败重新消费很正常,保证消息尽可能消费成功。

对,这句话不错,的确可以在一定程度上保证消费异常的消息可以消费成功。

但是坑不在这,而是前面提到的消费时传入的整个集合 中的消息都需要被重新消费。

具体的原因我们接着往下看

当消息处理之后,不论是成功还是异常,都需要对结果进行处理,代码如下

5809f58a-eb16-11ed-90ce-dac502259ad0.png

当处理结果为RECONSUME_LATER的时候(异常会设置为RECONSUME_LATER),此时ackIndex会设置成-1,后面循环遍历的时候就会遍历到所有这次消费的消息,然后调用sendMessageBack方法,sendMessageBack方式是用来实现消息重新消费的逻辑,这里就不展开说了。

所以,一旦被消费的一批消息中出现一个消费异常的情况,那么就会导致整批消息被重新消费,从而会导致在出现异常之前的成功处理的消息都会被重复消费 ,非常坑。

不过好在消费时传入的消息集合中的消息数量是可以设置的,并且默认就是1

58362cea-eb16-11ed-90ce-dac502259ad0.png

也就说默认情况下那个集合中就一条消息,所以默认情况下不会出现消费成功的消息被重复消费的情况。

所以这个参数不要轻易设置,一旦设置大了,就可能导致消息被重新消费。

除了并发消费消息的模式以外,RocketMQ还支持顺序消费消息的模式,也会造成重复消费,逻辑其实差不多,但是在实现消息重新消费的逻辑不一样。

消费者提交offset失败

首先来讲一讲什么是offset。

前面说过,消息在发送的时候需要指定发送到,消息最后会被放到Queue中,其实真正的消息不是在Queue中,Queue存的是每个消息的位置,但是你可以理解为Queue存的是消息。

而消息在Queue中是有序号的,这个序号就被称为offset,从0开始,单调递增1。

58555066-eb16-11ed-90ce-dac502259ad0.png

比如说,如上图,消息1的offset就是0,消息2的offset就是1,依次类推。

这个offset的一个作用就是用来管理消费者的消费进度。

当消费者在成功消费消息之后,需要将所消费的消息的offset提交给RocketMQ服务端,告诉RocketMQ,这个Queue的消息我已经消费到了这个位置了。

提交offset的代码就在上述第二节提到的处理结果的后面

58625cde-eb16-11ed-90ce-dac502259ad0.png

这样有一个好处,那么一旦消费者重启了或者其它啥的要从这个Queue拉取消息的时候,此时他只需要问问RocketMQ服务端上次这个Queue消息消费到哪个位置了,之后消费者只需要从这个位置开始消费消息就行了,这样就解决了接着消费的问题。

但是RocketMQ在设计的时候,当消费完消息的时候并不是同步告诉RocketMQ服务端offset,而是定时发送。

5887c3e8-eb16-11ed-90ce-dac502259ad0.png

如图,当消费者消费完消息的时候,会将offset保存到内存中的一个Map数据结构中,所以上面截图的那段代码其实是更新内存中的offset

589c957a-eb16-11ed-90ce-dac502259ad0.png

而在消费者启动的时候会开启一个定时任务,默认是5s一次,会通过网络请求将内存中的每个Queue的消费进度offset发送给RocketMQ服务端。

58b27b74-eb16-11ed-90ce-dac502259ad0.png

由于是定时任务,所以就可能出现服务器一旦宕机,导致最新消费的offset没有成功告诉RocketMQ服务端的情况

此时,消费进度offset就丢了,那么消费者重启的时候只能从RocketMQ中获取到上一次提交的offset,从这里开始消费,而不是最新的offset,出现明明消费到了第8个消息,RocketMQ却告诉他只消费到了第5个消息的情况,此时必然会导致消息又出现重复消费 的情况。

服务端持久化offset失败

上一节说到,消费者会有一个每隔5s钟的定时任务将每个队列的消费进度offset提交到RocketMQ服务端

当RocketMQ服务端接收到提交请求之后,会将这个消费进度offset保存到内存中

58d0a072-eb16-11ed-90ce-dac502259ad0.png

同时为了保证RocketMQ服务端重启消费进度不会丢失,也会开启一个定时任务,默认也是5s一次,将内存中的消费进度持久化到磁盘文件中

58f805ea-eb16-11ed-90ce-dac502259ad0.png

所以,整个消费进度offset的数据流转过程如下

5906fe7e-eb16-11ed-90ce-dac502259ad0.png

当RocketMQ服务端重启之后,会从磁盘中读取文件的数据加载到内存中。

跟消费者产生的问题一样,一旦RocketMQ发生宕机,那么offset就有可能丢失5s钟的数据,RocketMQ服务端一旦重启,消费者从RocketMQ服务端获取到的消息消费进度就比实际消费的进度低,同样也会导致消息重复消费。

主从同步offset失败

在RocketMQ的高可用模式中,有一种名叫主从同步的模式,当主节点挂了之后,从节点可以手动升级为主节点对外提供访问,保证高可用。

在主从同步模式下,从节点默认每隔10s会向主节点发送请求,同步一些元数据,这些元数据就包括消费进度

591ebd98-eb16-11ed-90ce-dac502259ad0.png

当从节点获取到主节点的消费进度之后,会将主节点的消费进度设置到自己的内存中,同时也会持久化到磁盘。

所以整个消费进度offset的数据的流转过程就会变成如下

59417978-eb16-11ed-90ce-dac502259ad0.png

同样,由于也是定时任务,那么一旦主节点挂了,从节点就会丢10s钟的消费进度,此时如果从节点升级为主节点对外提供访问,就会出现跟上面提到的一样的情况,消费者从这个新的主节点中拿到的消费进度比实际的低,自然而然就会重复消费消息。

所以,总的来说,在消费进度数据流转的过程中,只要某个环节出现了问题,都有很有可能会导致消息重复消费。

重平衡

先来讲一讲什么是重平衡,其实重平衡很好理解,我说一下你就明白了。

前面说到,消费者是从队列中获取消息的

59727820-eb16-11ed-90ce-dac502259ad0.png

在RocketMQ中,有个消费者组的概念,一个消费者组中可以有多个消费者,不同消费者组之间消费消息是互不干扰的,所以前面提到的消费者其实都在消费组下

5988424a-eb16-11ed-90ce-dac502259ad0.png

在同一个消费者组中,消息消费有两种模式:

集群消费模式

广播消费模式

由于RocketMQ默认 是集群消费模式,并且绝大多数业务场景都是使用集群消费模式,所以这里就不讨论广播消费模式了。

集群消费模式 是指同一条消息只能被这个消费者组消费一次,这就叫集群消费。

并且前面提到提交消费进度给RocketMQ服务端的情况只会集群消费模式下才会有,在广播消费模式不会提给到RocketMQ服务端,仅仅持久化到本地磁盘

同时前面说的消费者提交消费进度真正提交的是消费者组对于这个Queue的消费进度,而不是指具体的某个消费者对于Queue消费进度。

虽然说这里将前面提到的一些含义更深一步,但是并不妨碍前面的理解。

集群消费的实现就是将队列按照一定的算法分配给消费者,默认是按照平均分配的。

5998a216-eb16-11ed-90ce-dac502259ad0.png

如图所示,假设某个topic有4个Queue,有个消费者组订阅了这个topic,这个消费者组有两个消费者1和消费者2,此时每个消费者就可以被分配两个队列,这样就能保证消息正常情况下只会被消费一次。如果只有一个消费者,那么这个消费者就会消费所有队列,很好理解。

接着后面又启动了一个消费者3,此时为了保证刚上线的消费者3能够消费消息,就要进行重平衡 操作,重新分配每个消费者消费的队列。

在重平衡之后就可能会出现下面这种情况

59a98f36-eb16-11ed-90ce-dac502259ad0.png

如上图,原本被消费者2消费的Queue4被分配给消费者3,此时消费者3就能消费到消息了,这就是重平衡

除了新增消费者会导致重平衡之外,消费者数量减少,队列的数量增加或者减少都会触发重平衡。

在了解了重平衡概念之后,接下来分析一下为什么重平衡会导致消息的重复消费。

假设在进行重平衡时,还未重平衡完之前,消费者2此时还是会按照上面第二节提到的消费消息的逻辑来消费Queue4的消息

当消费者2已经重平衡完成了,发现Queue4自己已经不能消费了,那么此时就会把这个Queue4设置为dropped,就是丢弃的意思

59bb4f46-eb16-11ed-90ce-dac502259ad0.png

但是由于重平衡进行时消费者2仍然在消费Queue4的消息,但是当消费完之后,发现队列被设置成dropped,那么此时被消费者2消费消息的offset就不会被提交 ,原因如下代码

59e5afac-eb16-11ed-90ce-dac502259ad0.png

这段代码前面已经出现过,一旦dropped被设置成true,这个if条件就通不过,消费进度就不会被提交。

成功消费消息了,但是却不提交消费进度,这就非常坑了。。

于是当消费者3开始消费Queue4的消息的时候,他就会问问RocketMQ服务端,我消费者3所在的消费者组对于Queue4这个队列消费到哪了,我接着消费就行了。

此时由于没有提交消费进度,RocketMQ服务端告诉消费者3的消费进度就会比实际的低,这就造成了消息重复消费的情况。

清理长时间消费的消息

在RocketMQ中有这么一个机制,会定时清理长时间正在消费的消息。

5a0db4c0-eb16-11ed-90ce-dac502259ad0.png

如图,假设有5条消息现在正在被消费者处理,这5条消息会被存在一个集合中,并且是按照offset的大小排序,消息1的offset最小,消息5的offset最大。

RocketMQ消费者启动时会开启一个默认15分钟执行一次的定时任务

5a1d61d6-eb16-11ed-90ce-dac502259ad0.png

这个定时任务会去检查正在处理的消息的第一条消息,也就是图中的消息1,一旦发现消息1已经处理了超过15分钟了,那么此时就会将消息1从集合中移除,之后会隔一定时间再次消费消息1。

这也会有坑,虽然消息1从集合中被移除了,但是消息1并没有消失,仍然被消费者继续处理,但是消息1隔一定时间就会再次被消费,就会出现消息1被重复消费的情况。

这就是清理长时间消费的消息导致重复消费的原因。

但此时又会引出一个新的疑问,为什么要移除这个处理超过15分钟的消息呢?

这就又跟前面提到的消费进度 提交有关!

前面说过消息被消费完成之后会提交消费进度,提交的消费进度实际会有两种情况:

第一种 就是某个线程消费了所有的消息,当把所有的消息都消费完成之后,就会把消息从集合中全部移除,此时提交的消费进度offset就是图中消息5的offset+1

加1的操作是为了保证如果发生重启,那么消费者下次消费的起始位置就是消息5后面的消息,保证消息5不被重复消费

第二种 情况就不太一样了

假设现在有两个线程来处理这5条消息,线程1处理前2条,线程2处理后3条,如图

5a36b1ae-eb16-11ed-90ce-dac502259ad0.png

现在线程1出现了长时间处理消息的情况。

此时线程2处理完消息之后,移除后面三条消息,准备提交offset的时候发现集合中还有元素,就是线程1正在处理的前两条消息,此时线程2提交的offset并不是消息5对应的offset,而是消息1的offset,代码如下

5a42f22a-eb16-11ed-90ce-dac502259ad0.png

这么做的主要原因就是保证消息1和消息2至少被消费一次。

因为一旦提交了消息5对应的offset,如果消费者重启了,下次消费就会接着从消息5的后面开始消费,而对于消息1和消息2来说,并不知道有没有被消费成功,就有可能出现消息丢失的情况。

所以,一旦集合中最前面的消息长时间处理,那么就会导致后面被消费的消息进度无法提交,那么重启之后就会导致大量消息被重复消费。

为了解决这个问题,RocketMQ引入了定时清理的机制,定时清理长时间消费的消息,这样消费进度就可以提交了。

最后

总得来说,RocketMQ中还是存在很多种导致消息重读消费的情况,并且官方也说了,只是在大多数情况下消息不会重复

5a82f546-eb16-11ed-90ce-dac502259ad0.png

所以如果你的业务场景中需要保证消息不能重复消费,那么就需要根据业务场景合理的设计幂等技术方案。





审核编辑:刘清

声明:本文内容及配图由入驻作者撰写或者入驻合作网站授权转载。文章观点仅代表作者本人,不代表电子发烧友网立场。文章及其配图仅供工程师学习之用,如有内容侵权或者其他违规问题,请联系本站处理。 举报投诉
  • 服务器
    +关注

    关注

    12

    文章

    9129

    浏览量

    85350
  • Queue
    +关注

    关注

    0

    文章

    16

    浏览量

    7261
  • null
    +关注

    关注

    0

    文章

    18

    浏览量

    3968

原文标题:RocketMQ源码中,7种导致重复消费的坑!

文章出处:【微信号:芋道源码,微信公众号:芋道源码】欢迎添加关注!文章转载请注明出处。

收藏 人收藏

    评论

    相关推荐

    与无线网络相关的那些事

    ,体验上网带来的愉悦(ps:天下没有免费的午餐,认证登陆界面的广告以及商户提供产品内容,也能接受这样的广告植入形式)在上菜时,而不是先吃而是拍照片发微博\微信到朋友圈内,这也间接的为商户进行口碑宣传。你与无线网络都有哪些趣
    发表于 05-27 11:40

    我看过的那些Linux相关的书籍

    来北京工作已经个多月,大都市的生活比起读大学要忙碌得多,尤其是出行,基本以小时为基本的计时单位。有时茫然看着窗外车水马龙,会有些迷茫自己选择的是对还是错。  题外话不多说,回归这次的主题,
    发表于 07-04 06:39

    渣机产品有哪些参数

     铣挖机履带式挖掘装载机(俗名:渣机)是种连续生产的高效率出矿设备,主要用于矿山岩巷、半煤岩巷掘进,也可以用于引水洞、铁路隧道施工和国防洞窟施工中的装载作业,渣机与我公司生产的煤矿用液压钻车
    发表于 09-02 08:00

    电源模块发热的原因

    次,我们引起电源模块发热的原因。电源模块在电压转换过程中有能量损耗,产生热能导致模块发热,降低电源的转换效率,影响电源模块正常工作,并
    发表于 01-03 07:38

    C语言hello world背后的内幕

    行时,它在内存中是什么样子的?程序的执行入口为什么是 main 函数?可执行文件的内部结构是怎么样的?闲话少说,让我们进入正题, hello world 背后的内幕。注:本文是在 Ubuntu
    发表于 09-30 10:31

    美容仪哪个牌子好?来令人眼花缭乱的日本美容仪

    获得了大众的喜爱。美容仪哪个牌子好?小编给你日本的美容仪神器品牌。 我们都知道,日本是个科技大国,不管是电器类还是美容仪类,都收获了世界大批粉丝的追捧,市面上的美容仪品牌多种多样,价格相差也大,各大美容仪从洗脸到瘦脸
    发表于 04-16 19:50 9852次阅读
    美容仪哪个牌子好?来<b class='flag-5'>扒</b><b class='flag-5'>一</b><b class='flag-5'>扒</b>令人眼花缭乱的日本美容仪

    好用的日本家用美容仪品牌,让你享受清洁肌肤的乐趣

    了,很多人都愿意尝试美容仪带来的护肤体验。美容仪真的有用吗?它的价格相比较于去美容院会划算很多,但是却比般的护肤保养品昂贵,是真的物有所值还是商家的噱头?今天就来日本好用的家用
    发表于 06-04 21:03 854次阅读

    店saas系统创新性服务平台的优势是什么

    也在不断升级换代,比如目前新代的店智能数字店铺系统。 前沿科技让店铺数字化管理趋于精准,店数字店铺系统除了常规的收银支付等最基本的功能外,结合当前最前沿的人工智能、5G、大数据等技术于
    的头像 发表于 10-15 10:14 2035次阅读

    中断为什么不能调printf?

    前面说会写下Modbus-RTU的实现,写了1000多字了,有兴趣的稍等下哈。前面在个群里看到个朋友在个串口接收中断里打印遇到了问
    发表于 12-04 12:21 0次下载
    <b class='flag-5'>扒</b><b class='flag-5'>一</b><b class='flag-5'>扒</b>中断为什么不能调printf?

    个超棒的stm32的开源usb-can项目,canable及PCAN固件

    个超棒的stm32的开源usb-can项目,canable及PCAN固件
    发表于 12-20 18:55 36次下载
    <b class='flag-5'>扒</b><b class='flag-5'>一</b>个超棒的stm32的开源usb-can项目,canable及PCAN固件

    RocketMQ中各类重复消费的原理浅析

    利用消息中间件,如何保证MQ消费消息的幂等性?所谓知其然,才能知其所以然,本文将通过RocketMQ作为例子,来什么情况下会导致
    的头像 发表于 01-08 09:29 1156次阅读
    <b class='flag-5'>RocketMQ</b>中各类<b class='flag-5'>重复</b>消费的原理浅析

    雕铣机、雕刻机和加工中心之间的区别

    中心、雕铣机、雕刻机,之间有什么区别?相信这句话很多刚刚加入这个圈的朋友都会问,然后在买机械设备的时候不太懂,不知道怎么区分,到底应该买什么样的设备,才能达到自己的需求,今天小编就为大家他们三者之间的区别。
    的头像 发表于 01-15 09:48 759次阅读

    晶振频率漂移的原因

    晶振频率漂移的原因  晶振频率漂移是指晶振器输出频率在长时间使用中逐渐偏离其标称频率的现象。晶振频率漂移是种晶振器的固有性能,其
    的头像 发表于 01-26 14:20 1004次阅读

    折叠屏手机背后的“黑科技”

    折叠屏手机似乎正成为各大品牌下步推新的产品,那么今天,就让我们这里面到底有什么“黑科技”。
    的头像 发表于 02-26 10:34 1290次阅读
    <b class='flag-5'>扒</b><b class='flag-5'>一</b><b class='flag-5'>扒</b>折叠屏手机背后的“黑科技”

    渣机远程监控运维管理系统解决方案

    渣机主要由机械手与输送机相结合,将自动渣和自动输送功能合二为,替代了传统的人工和间歇式机械作业,大大提高了工作效率,降低了劳动强度,并显著提升了作业安全性,在矿山、道路、隧道、水利等工程领域
    的头像 发表于 09-05 17:10 190次阅读