当前位置：首页 > news >正文

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

news 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 从小到大顺序获取。