rocketmq 单消费者多线程消费是如何保证分区内的顺序消息严格按照 FIFO 顺序消费的想要实现真正的顺序消费必须同时实现生产顺序性和消费顺序性。生产顺序性采用单一生产者单线程串行生产消息。或者保证同一个分区键的消息能交由同一个生产者串行发送。消费顺序性保证同一个分区键的消息按照投递顺序串行消费。生产顺序性比较好实现消费顺序性如果采用单线程消费也好实现但是消费的速度太慢了很容易导致消息积压。rocketmq 默认单个消费者是多线程消费的我之前一直好奇多线程消费是如何保证同一个分区键的消息按照投递顺序串行消费的最近研究了一下。这里基于 rocketmq 4.x 版本说明rocketmq消费模式时分为并发消费和顺序消费启动消费者时需要设置MessageListener实现具体的消费逻辑并发消费使用MessageListenerConcurrently没有任何顺序保证顺序消费使用MessageListenerOrderly能保证同一个分区键的消息按照投递顺序串行消费默认的消费线程数都为 20。消息模式有两种广播模式下消息会被广播给消费组内的每一个消费者集群模式下 rocketmq 能保证一个分区rocketmq 称为队列的消息只会被同一消费组内的一个消费者消费。使用DefaultMQPushConsumer推模式测试客户端从 MQ 中获取消息后会将消息提交到线程池中执行。topic 的默认分区数队列数是 8queueId 是 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 为 offsetvalue 为MessageExt在takeMessages时能保证按照 offset 从小到大顺序获取。在 433 行对messageQueue加了synchronized锁messageQueue代表的是主题 topic_order2 的队列 1。437 行 for 循环一直消费直到消费失败或者没有可消费的消息了。如果因为消费失败而退出循序在下一次执行run方法会自动重试失败的消息。466 行ListMessageExt msgs this.processQueue.takeMessages(consumeBatchSize)获取消息。490 到 580 行获取processQueue的ConsumeLock加锁并消费、解锁。总结顺序消费利用messageQueueLock和processQueueConsumeLock两个锁将同一 topic 的同一 queue 的消息的消费给串行化了。由于 queue 的顺序消费是串行化的所以在创建 topic 时一个尽量将分区队列数设置的大一点否则容易消息积压。
rocketmq 单消费者多线程消费是如何保证分区内的顺序消息严格按照 FIFO 顺序消费的？
rocketmq 单消费者多线程消费是如何保证分区内的顺序消息严格按照 FIFO 顺序消费的想要实现真正的顺序消费必须同时实现生产顺序性和消费顺序性。生产顺序性采用单一生产者单线程串行生产消息。或者保证同一个分区键的消息能交由同一个生产者串行发送。消费顺序性保证同一个分区键的消息按照投递顺序串行消费。生产顺序性比较好实现消费顺序性如果采用单线程消费也好实现但是消费的速度太慢了很容易导致消息积压。rocketmq 默认单个消费者是多线程消费的我之前一直好奇多线程消费是如何保证同一个分区键的消息按照投递顺序串行消费的最近研究了一下。这里基于 rocketmq 4.x 版本说明rocketmq消费模式时分为并发消费和顺序消费启动消费者时需要设置MessageListener实现具体的消费逻辑并发消费使用MessageListenerConcurrently没有任何顺序保证顺序消费使用MessageListenerOrderly能保证同一个分区键的消息按照投递顺序串行消费默认的消费线程数都为 20。消息模式有两种广播模式下消息会被广播给消费组内的每一个消费者集群模式下 rocketmq 能保证一个分区rocketmq 称为队列的消息只会被同一消费组内的一个消费者消费。使用DefaultMQPushConsumer推模式测试客户端从 MQ 中获取消息后会将消息提交到线程池中执行。topic 的默认分区数队列数是 8queueId 是 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 为 offsetvalue 为MessageExt在takeMessages时能保证按照 offset 从小到大顺序获取。在 433 行对messageQueue加了synchronized锁messageQueue代表的是主题 topic_order2 的队列 1。437 行 for 循环一直消费直到消费失败或者没有可消费的消息了。如果因为消费失败而退出循序在下一次执行run方法会自动重试失败的消息。466 行ListMessageExt msgs this.processQueue.takeMessages(consumeBatchSize)获取消息。490 到 580 行获取processQueue的ConsumeLock加锁并消费、解锁。总结顺序消费利用messageQueueLock和processQueueConsumeLock两个锁将同一 topic 的同一 queue 的消息的消费给串行化了。由于 queue 的顺序消费是串行化的所以在创建 topic 时一个尽量将分区队列数设置的大一点否则容易消息积压。
