尧图网站建设 尧图网络
  • 首页
  • 关于我们
  • 服务项目
  • 案例展示
  • 建站流程
  • 资讯中心
  • 联系我们
首页/资讯中心/详情

rocketmq 单消费者多线程消费是如何保证分区内的顺序消息严格按照 FIFO 顺序消费的?

rocketmq 单消费者多线程消费是如何保证分区内的顺序消息严格按照 FIFO 顺序消费的?
📅 发布时间:2026/7/9 17:12:47

rocketmq 单消费者多线程消费是如何保证分区内的顺序消息严格按照 FIFO 顺序消费的?

想要实现真正的顺序消费,必须同时实现生产顺序性和消费顺序性。

生产顺序性:采用单一生产者单线程串行生产消息。或者保证同一个分区键的消息能交由同一个生产者串行发送。

消费顺序性:保证同一个分区键的消息按照投递顺序串行消费。

生产顺序性比较好实现,消费顺序性如果采用单线程消费也好实现,但是消费的速度太慢了,很容易导致消息积压。rocketmq 默认单个消费者是多线程消费的,我之前一直好奇多线程消费是如何保证同一个分区键的消息按照投递顺序串行消费的,最近研究了一下。

这里基于 rocketmq 4.x 版本说明,rocketmq消费模式时分为并发消费和顺序消费,启动消费者时需要设置MessageListener实现具体的消费逻辑,并发消费使用MessageListenerConcurrently,没有任何顺序保证;顺序消费使用MessageListenerOrderly,能保证同一个分区键的消息按照投递顺序串行消费,默认的消费线程数都为 20。

消息模式有两种,广播模式下消息会被广播给消费组内的每一个消费者,集群模式下 rocketmq 能保证一个分区(rocketmq 称为队列)的消息只会被同一消费组内的一个消费者消费。

使用DefaultMQPushConsumer推模式测试,客户端从 MQ 中获取消息后,会将消息提交到线程池中执行。

topic 的默认分区数(队列数)是 8,queueId 是 0 到 7。

从 mq 获取消息

pullResult是消费组从 mq 获取消息的请求,下图中是消费者组 order_c2 从主题 topic_order2 的队列 1 偏移量是 201 的地方获取消息,获取成功后 338 行先将获取的消息列表放入processQueue中,然后调用DefaultMQPushConsumerImpl.this.consumeMessageService.submitConsumeRequest向线程池提交任务。

并发消费

并发消费由org.apache.rocketmq.client.impl.consumer.ConsumeMessageConcurrentlyService#submitConsumeRequest方法提交任务,任务对象为org.apache.rocketmq.client.impl.consumer.ConsumeMessageConcurrentlyService.ConsumeRequest,构造方法有三个参数new ConsumeRequest(msgs, processQueue, messageQueue)。

以下是ConsumeRequest#run方法的实现,被选中的status = listener.consumeMessage(Collections.unmodifiableList(msgs), context);就是调用MessageListenerConcurrently#consumeMessage方法执行自定义的消费逻辑,前面没有任何的加锁或其他同步代码,所以没有任何顺序保证。消费完构造方法传入的msgs就结束了。

顺序消费

顺序消费由org.apache.rocketmq.client.impl.consumer.ConsumeMessageOrderlyService#submitConsumeRequest方法提交任务,任务对象为org.apache.rocketmq.client.impl.consumer.ConsumeMessageOrderlyService.ConsumeRequest,构造方法只有两个参数new ConsumeRequest(processQueue, messageQueue)。

messageQueue与processQueue的作用、区别:

messageQueue只有topic、brokerName、queueId三个字段,是 mq 分区(队列)的一个简单代表。

processQueue中维护了messageQueue所代表的分区(队列)的待消费的消息、offset、数量等详细的信息。

processQueue使用TreeMap维护消息,key 为 offset,value 为MessageExt,在takeMessages时能保证按照 offset 从小到大顺序获取。

在 433 行对messageQueue加了synchronized锁,messageQueue代表的是主题 topic_order2 的队列 1。

437 行 for 循环一直消费,直到消费失败或者没有可消费的消息了。如果因为消费失败而退出循序,在下一次执行run方法会自动重试失败的消息。

466 行List<MessageExt> msgs = this.processQueue.takeMessages(consumeBatchSize)获取消息。

490 到 580 行,获取processQueue的ConsumeLock加锁并消费、解锁。

总结:顺序消费利用messageQueueLock和processQueueConsumeLock两个锁将同一 topic 的同一 queue 的消息的消费给串行化了。

由于 queue 的顺序消费是串行化的,所以在创建 topic 时一个尽量将分区(队列)数设置的大一点,否则容易消息积压。

相关新闻

  • 亨得利官方名表服务中心|最新官方热线及详细地址权威信息公示(2026年7月最新) - 亨得利钟表维修中心
  • 基于TPD2015FN与STM32的高性能负载控制系统设计
  • Claude Code 是什么?深度解析社区驱动的 Claude CLI 工具链

最新新闻

  • 2026免费图片转文字工具全解:在线,OCR,网站、手机端、小程序实操指南
  • Conda 环境管理:Python 3.10.12 项目部署的 5 个关键步骤与 3 个常见误区
  • 2026劳力士回收靠谱平台测评|南京五家名表机构分级对比指南 - 每日生活报
  • Three.js 波涛海浪教程
  • Claude Code 实战:AI 结对编程如何真正提效-0620
  • JetBrains 推面向团队的 AI 系统,统一开发流程并升级商业模型

日新闻

  • SQL 查询语句的标准逻辑执行顺序(即语义处理顺序),它与实际书写顺序不同,但决定了数据库如何解析和执行查询
  • ORB-SLAM2 重定位模块深度解析:从 BoW 候选帧到 PnP 优化的 6 步流程
  • 罗技鼠标宏压枪脚本终极指南:从原理到实战的完整解析

周新闻

  • 基于YOLOv12的番茄成熟度智能检测系统开发
  • 终极RimWorld模组管理指南:用RimSort告别模组冲突烦恼
  • AI Agent框架开发:从理论到实践的完整指南

月新闻

  • 2026年6月公司网站搭建最新热门渠道测评:四大低成本/零代码平台对比+避坑
  • 【Linux】Linux arm 编译QT程序,出现expected “}“报错
  • 【MATLAB例程】四基站二维AOA定位与距离辅助增强对比仿真。基于角度观测和测距修正的固定目标平面定位精度分析

关于尧图

  • 公司简介
  • 团队介绍
  • 企业文化
  • 荣誉资质

服务项目

  • 定制开发
  • 电商建站
  • UI 设计
  • 运维服务

快速链接

  • 案例展示
  • 建站流程
  • 常见问题
  • 资讯中心

联系方式

  • 📍北京市朝阳区互联网产业园 A 座 10 层
  • 📞400-888-8888
  • ✉️contact@rkmt.cn
  • 🕐周一至周日 9:00-21:00

© 2024 北京尧图网络科技有限公司 版权所有 | 京 ICP 备 XXXXXXXX 号