RocketMQ 一行代码造成大量消息丢失

简介: RocketMQ 一行代码造成大量消息丢失

1、问题现象


首先接到项目反馈使用 RocketMQ 会出现如下错误:

898fcbf9ff3f520e8242d86df73a1a05.png

错误信息关键点:MQBrokerException:CODE:2 DESC:[TIMEOUT_CLEAN_QUEUE]broker busy,start flow control for a while,period in queue:205ms,size of queue:880。


由于项目组并没有对消息发送失败做任何补偿,导致丢失消息发送失败,故需要对这个问题进行深层次的探讨,并加以解决。


2、问题分析


首先我们根据关键字:TIMEOUT_CLEAN_QUEUE 去 RocketMQ 中查询,去探究在什么时候会抛出如上错误。根据全文搜索如下图所示:


94d1dc6a8d87ddce3309ff8234ce96c8.png

该方法是在 BrokerFastFailure 中定义的,通过名称即可以看成其设计目的:Broker端快速失败机制。


Broker 端快速失败其原理图如下:


1cb45a87842ea12efbb2aef2ac6727ee.png

  • 消息发送者向 Broker 发送消息写入请求,Broker 端在接收到请求后会首先放入一个队列中(SendThreadPoolQueue),默认容量为 10000。
  • Broker 会专门使用一个线程池(SendMessageExecutor)去从队列中获取任务并执行消息写入请求,为了保证消息的顺序处理,该线程池默认线程个数为1。


如果 Broker 端受到垃圾回收等等因素造成单条写入数据发生抖动,单个 Broker 端积压的请求太多从而得不到及时处理,会极大的造成客户端消息发送的时间延长。


设想一下,如果由于 Broker 压力增大,写入一条消息需要500ms甚至超过1s,并且队列中积压了5000条消息,消息发送端的默认超时时间为3s,如果按照这样的速度,这些请求在轮到 Broker 执行写入请求时,客户端已经将这个请求超时了,这样不仅会造成大量的无效处理,还会导致客户端发送超时。


故 RocketMQ 为了解决该问题,引入 Broker 端快速失败机制,即开启一个定时调度线程,每隔10毫秒去检查队列中的第一个排队节点,如果该节点的排队时间已经超过了 200ms,就会取消该队列中所有已超过 200ms 的请求,立即向客户端返回失败,这样客户端能尽快进行重试,因为 Broker 都是集群部署,下次重试可以发送到其他 Broker 上,这样能最大程度保证消息发送在默认 3s 的时间内经过重试机制,能有效避免某一台 Broker 由于瞬时压力大而造成的消息发送不可用,从而实现消息发送的高可用。


从 Broker 端快速失败机制引入的初衷来看,快速失败后会发起重试,除非同一时刻集群内所有的 Broker 都繁忙,不然消息会发送成功,用户是不会感知这个错误的,那为什么用户感知了呢?难道 TIMEOUT_ CLEAN _ QUEUE 错误,Broker 不重试?


为了解开这个谜团,接下来会采用源码分析的手段去探究真相。接下来将以消息同步发送为例揭示其消息发送处理流程中的核心关键点。


MQ Client 消息发送端首先会利用网络通道将请求发送到 Broker,然后接收到请求结果后并调用 processSendResponse 方法对响应结果进行解析,如下图所示:

16d9048d432abbc895534b6d16ee3aab.png

在这里返回的 code 为 RemotingSysResponseCode . SYSTEM_BUSY。

我们从 proccessSendResponse 方法中可以得知如果 code 为 SYSTEM_BUSY,该方法会抛出 MQBrokerException,响应 code 为 SYSTEM_BUSY,其错误描述为开头部分的错误信息。


那我们沿着该方法的调用链路,可以找到其直接调用方:DefaultMQProducerImpl 的 sendKernelImpl,我们重点考虑如果底层方法抛出  MQBrokerException 该方法会如何处理。


其关键代码如下图所示:

a74210e249d9c3241eca2efd64b28b1a.png

可以看出在 sendKernelImpl 方法中首先会捕捉异常,先执行注册的钩子函数,即就算执行失败,对应的消息发送后置钩子函数也会执行,然后再原封不动的将该异常向上抛出。


sendKernelImpl 方法被 DefaultMQProducerImpl 的 sendDefaultImpl 方法调用,下面是其核心实现截图:

dc658ccb6a703925f9e433f0701fe2b3.png

从这里可以看出 RocketMQ 消息发送高可用设计一个非常关键的点,重试机制,其实现是在 for 循环中 使用 try catch 将 sendKernelImpl 方法包裹,就可以保证该方法抛出异常后能继续重试。从上文可知,如果 SYSTEM_BUSY 会抛出 MQBrokerException,但发现只有上述几个错误码才会重试,因为如果不是上述错误码,会继续向外抛出异常,此时 for 循环会被中断,即不会重试。


这里非常令人意外的是连 SYSTEM_ERROR 都会重试,却没有包含 SYSTEM_BUSY,显然违背了快速失败的设计初衷,故笔者断定,这是 RocketMQ 的一个BUG,将 SYSTEM_BUSY 遗漏了,后续会提一个 PR,增加一行代码,将 SYSTEM_BUSY 加上即可。


问题分析到这里,该问题应该就非常明了。


3、解决方案


如果大家在网上搜索 TIMEOUT_CLEAN_QUEUE 的解决方法,大家不约而同提出的解决方案是增加 waitTimeMillsInSendQueue 的值,该值默认为 200ms,例如将其设置为 1000s 等等,以前我是反对的,因为我的认知里 Broker 会重试,但现在发现 Broker 不会重试,所以我现在认为该 BUG未解决的情况下适当提高该值能有效的缓解。


但这是并不是好的解决方案,我会在近期向官方提交一个PR,将这个问题修复,建议大家在公司尽量对自己使用的版本进行修改,重新打一个包即可,因为这已经违背了 Broker 端快速失败的设计初衷。


但在消息发送的业务方,尽量自己实现消息的重试机制,即不依赖 RocketMQ 本身提供的重试机制,因为受制于网络等因素,消息发送不可能百分之百成功,建议大家在消息发送时捕获一下异常,如果发送失败,可以将消息存入数据库,再结合定时任务对消息进行重试,尽最大程度保证消息不丢失。


相关实践学习
消息队列RocketMQ版:基础消息收发功能体验
本实验场景介绍消息队列RocketMQ版的基础消息收发功能,涵盖实例创建、Topic、Group资源创建以及消息收发体验等基础功能模块。
消息队列 MNS 入门课程
1、消息队列MNS简介 本节课介绍消息队列的MNS的基础概念 2、消息队列MNS特性 本节课介绍消息队列的MNS的主要特性 3、MNS的最佳实践及场景应用 本节课介绍消息队列的MNS的最佳实践及场景应用案例 4、手把手系列:消息队列MNS实操讲 本节课介绍消息队列的MNS的实际操作演示 5、动手实验:基于MNS,0基础轻松构建 Web Client 本节课带您一起基于MNS,0基础轻松构建 Web Client
相关文章
|
监控 物联网 API
【.NET+MQTT】.NET6 环境下实现MQTT通信,以及服务端、客户端的双边消息订阅与发布的代码演示
MQTT广泛应用于工业物联网、智能家居、各类智能制造或各类自动化场景等。MQTT是一个基于客户端-服务器的消息发布/订阅传输协议,在很多受限的环境下,比如说机器与机器通信、机器与物联网通信等。好了,科普的废话不多说,下面直接通过.NET环境来实现一套MQTT通信demo,实现服务端与客户端的双边消息发布与订阅的功能和演示。
1326 0
【.NET+MQTT】.NET6 环境下实现MQTT通信,以及服务端、客户端的双边消息订阅与发布的代码演示
|
5月前
|
物联网 C# Windows
看看如何使用 C# 代码让 MQTT 进行完美通信
看看如何使用 C# 代码让 MQTT 进行完美通信
680 0
|
5月前
|
Java
MQTT(EMQX) - Java 调用 MQTT Demo 代码
MQTT(EMQX) - Java 调用 MQTT Demo 代码
203 0
MQTT(EMQX) - Java 调用 MQTT Demo 代码
|
6月前
|
消息中间件 安全 PHP
消息队列 MQ使用问题之如何获取PHP客户端代码
消息队列(MQ)是一种用于异步通信和解耦的应用程序间消息传递的服务,广泛应用于分布式系统中。针对不同的MQ产品,如阿里云的RocketMQ、RabbitMQ等,它们在实现上述场景时可能会有不同的特性和优势,比如RocketMQ强调高吞吐量、低延迟和高可用性,适合大规模分布式系统;而RabbitMQ则以其灵活的路由规则和丰富的协议支持受到青睐。下面是一些常见的消息队列MQ产品的使用场景合集,这些场景涵盖了多种行业和业务需求。
|
5月前
|
消息中间件 开发工具
【Azure Event Hub】原生应用中使用RabbitMQ,是否可以不改动代码的情况下直接转换为使用Event Hub呢?
【Azure Event Hub】原生应用中使用RabbitMQ,是否可以不改动代码的情况下直接转换为使用Event Hub呢?
|
消息中间件 Java Maven
消息中间件系列教程(07) -RabbitMQ -案例代码(点对点队列模式)
消息中间件系列教程(07) -RabbitMQ -案例代码(点对点队列模式)
81 1
|
消息中间件 安全 Go
动态订阅时 rocketmq-client-go 代码有map并发bug
动态订阅时 rocketmq-client-go 代码有map并发bug
71 2
|
消息中间件 Java Maven
消息中间件系列教程(11) -RabbitMQ -案例代码(通配符模式)
消息中间件系列教程(11) -RabbitMQ -案例代码(通配符模式)
94 0
|
消息中间件 Java Maven
消息中间件系列教程(10) -RabbitMQ -案例代码(路由模式)
消息中间件系列教程(10) -RabbitMQ -案例代码(路由模式)
88 0
|
消息中间件 存储 Java
消息中间件系列教程(09) -RabbitMQ -案例代码(发布订阅模式)
消息中间件系列教程(09) -RabbitMQ -案例代码(发布订阅模式)
71 0