专栏首页中间件兴趣圈RocketMQ 一行代码造成大量消息丢失

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

1、问题现象


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

错误信息关键点: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 中查询,去探究在什么时候会抛出如上错误。根据全文搜索如下图所示:

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

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

  • 消息发送者向 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 方法对响应结果进行解析,如下图所示:

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

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

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

其关键代码如下图所示:

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

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

从这里可以看出 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 本身提供的重试机制,因为受制于网络等因素,消息发送不可能百分之百成功,建议大家在消息发送时捕获一下异常,如果发送失败,可以将消息存入数据库,再结合定时任务对消息进行重试,尽最大程度保证消息不丢失。

本文分享自微信公众号 - 中间件兴趣圈(dingwpmz_zjj),作者:丁威

原文出处及转载信息见文内详细说明,如有侵权,请联系 yunjia_community@tencent.com 删除。

原始发表时间:2020-05-18

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

我来说两句

0 条评论
登录 后参与评论

相关文章

  • RocketMQ 消息丢失场景分析及如何解决!

    既然在项目中使用了MQ,那么就不可避免的需要考虑消息丢失问题。在一些涉及到了金钱交易的场景下,消息丢失还是很致命的。那么在RocketMQ中存在哪几种消息丢失的...

    架构师修炼
  • RocketMQ高阶业务问题及解决方案

    Broker 是负责存储消息的,怎么保证消息发送到Broker后,一定不会丢失呢?

    大王叫下
  • 回答面试官:如何保证消息不丢失

    对于这个技术点不知道大家掌握的如何了,消息队列现在应该是公司必备的技能之一了,无论是RabbitMQ还是rocketmq,或者支持大数量的kafka

    Java宝典
  • RabbitMQ都写了,RocketMQ怎么能落下?

    最近看到了我在Github上写的rabbitmq-examples陆续被人star了,就想着写个rocketmq-examples。对rabbitmq感兴趣的小...

    Java识堂
  • RocketMQ 设计原理与最佳实践

    RocketMQ 是一款开源的分布式消息系统,基于高可用分布式集群技术,提供低延时的、高可靠的消息发布与订阅服务。同时,广泛应用于多个领域,包括异步通信解耦、企...

    Java识堂
  • 爬虫架构 | 消息队列应用场景及ActiveMQ、RabbitMQ、RocketMQ、Kafka对比

    黄小怪
  • 还不知道事务消息吗?这篇文章带你全面扫盲!

    在分布式系统中,为了保证数据一致性是必须使用分布式事务。分布式事务实现方式就很多种,今天主要介绍一下使用 RocketMQ 事务消息,实现分布事务。

    andyxh
  • 一个天天用消息队列的人,不知道为啥用 MQ,这就有点尴尬

    分析:一个用消息队列的人,不知道为啥用,有点尴尬。没有复习这点,很容易被问蒙,然后就开始胡扯了。

    lyb-geek
  • 30分钟带你了解「消息中间件」Kafka、RocketMQ

    https://www.open-open.com/lib/view/open1421150566328.html

    Yano_nankai
  • 为什么要使用MQ消息中间件?这3个点让你彻底明白!

    一个用消息队列的人,不知道为啥用,有点尴尬。没有复习这点,很容易被问蒙,然后就开始胡扯了。

    程序员追风
  • 天天在用消息队列,却不知道为啥要用 MQ ,这就尴尬了

    一个用消息队列的人,不知道为啥用,有点尴尬。没有复习这点,很容易被问蒙,然后就开始胡扯了。

    芋道源码
  • 1000000 / 60S 的 RocketMQ 不停机,扩容,平滑升级!

    1、各业务系统持续迭代过程中,JDK、SpringBoot、RocketMQ Client 等框架也进行了升级,高版本的 RocketMQ Client 发送的...

    鲁强
  • RocketMQ与Kafka对比(18项差异)

    转自:https://github.com/alibaba/RocketMQ/wiki/rmq_vs_kafka 淘宝内部的交易系统使用了淘宝自主研发的Noti...

    Albert陈凯
  • 分布式之消息队列复习精讲

    Java高级架构
  • 分布式之消息队列复习精讲

    庆幸的是两位朋友都很有上进心,于是博主写这篇文章,帮助他们复习一下关于消息队列中间件这块的要点

    Java团长
  • 分布式之消息队列复习精讲

    庆幸的是两位朋友都很有上进心,于是博主写这篇文章,帮助他们复习一下关于消息队列中间件这块的要点

    lyb-geek
  • 分布式之消息队列复习精讲!

    小A,工作于传统软件行业(某社保局的软件外包公司),每天工作内容就是和产品聊聊需求,改改业务逻辑。再不然就是和运营聊聊天,写几个SQL,生成下报表。又或者接到客...

    Java后端技术
  • 让分布式消息队列不再难懂

    小A,工作于传统软件行业(某社保局的软件外包公司),每天工作内容就是和产品聊聊需求,改改业务逻辑。再不然就是和运营聊聊天,写几个SQL,生成下报表。又或者接到客...

    java思维导图
  • 分布式之消息队列复习精讲!

    小A,工作于传统软件行业(某社保局的软件外包公司),每天工作内容就是和产品聊聊需求,改改业务逻辑。再不然就是和运营聊聊天,写几个SQL,生成下报表。又或者接到客...

    用户5224393

扫码关注云+社区

领取腾讯云代金券