RocketMQ源码：顺序消费实现原理

╰半橙微兮° 2024-03-23 16:25 12阅读 0赞

## 1. 介绍 ##

**全局有序**  
在RocketMQ中，如果使消息全局有序，可以为Topic设置一个消息队列，使用一个生产者单线程发送数据，消费者端也使用单线程进行消费，从而保证消息的全局有序，但是这种方式效率低，一般不使用。

![format_png][]

**局部有序**  
假设一个Topic分配了两个消息队列，生产者在发送消息的时候，可以对消息设置一个路由ID，比如想保证一个订单的相关消息有序，那么就使用订单ID当做路由ID，在发送消息的时候，通过订单ID对消息队列的个数取余，根据取余结果选择消息队列，这样同一个订单的数据就可以保证发送到一个消息队列中，消费者端使用`MessageListenerOrderly`处理有序消息，这就是RocketMQ的局部有序，保证消息在某个消息队列中有序。

接下来看RoceketMQ源码中提供的顺序消息例子（稍微做了一些修改）：

**生产者**

public class Producer {
        public static void main(String[] args) throws UnsupportedEncodingException {
            try {
                // 创建生产者
                DefaultMQProducer producer = new DefaultMQProducer("生产者组");
                // 启动
                producer.start();
                // 创建TAG
                String[] tags = new String[] {"TagA", "TagB", "TagC", "TagD", "TagE"};
                for (int i = 0; i < 100; i++) {
                    // 生成订单ID
                    int orderId = i % 10;
                    // 创建消息
                    Message msg =
                        new Message("TopicTest", tags[i % tags.length], "KEY" + i,
                            ("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET));
                    SendResult sendResult = producer.send(msg, new MessageQueueSelector() {
                        @Override
                        public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
                            // 获取订单ID
                            Integer id = (Integer) arg;
                            // 对消息队列个数取余
                            int index = id % mqs.size();
                            // 根据取余结果选择消息要发送给哪个消息队列
                            return mqs.get(index);
                        }
                    }, orderId); // 这里传入了订单ID
                    System.out.printf("%s%n", sendResult);
                }
    
                producer.shutdown();
            } catch (MQClientException | RemotingException | MQBrokerException | InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
    复制代码

**消费者**

public class Consumer {
    
        public static void main(String[] args) throws MQClientException {
            // 创建消费者
            DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("消费者组");
            consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
            // 订阅主题
            consumer.subscribe("TopicTest", "TagA || TagC || TagD");
            // 注册消息监听器，使用的是MessageListenerOrderly
            consumer.registerMessageListener(new MessageListenerOrderly() {
                @Override
                public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
                    context.setAutoCommit(true);
                    // 打印消息
                    System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgs);
                    return ConsumeOrderlyStatus.SUCCESS;
                }
            });
            consumer.start();
            System.out.printf("Consumer Started.%n");
        }
    }
    复制代码

从例子中可以看出生产者在发送消息的时候，通过`订单ID作为路由信息`，将`同一个订单ID的消息发送到了同一个消息队列中`，保证同一个订单ID的消息有序，那么消费者端是如何保证消息的顺序读取呢？接下来就去看下源码。

## 2. 顺序消息实现原理 ##

在[RocketMQ源码16-consumer消费流程][RocketMQ_16-consumer]一文中讲到，消费者在启动时会调用`DefaultMQPushConsumerImpl`的start方法：

public class DefaultMQPushConsumer extends ClientConfig implements MQPushConsumer {
        
        /**
         * 默认的消息推送实现类
         */
        protected final transient DefaultMQPushConsumerImpl defaultMQPushConsumerImpl;
        
        /**
         * 启动
         */
        @Override
        public void start() throws MQClientException {
            setConsumerGroup(NamespaceUtil.wrapNamespace(this.getNamespace(), this.consumerGroup));
            // 启动消费者
            this.defaultMQPushConsumerImpl.start();
            // ...
        }
    }
    复制代码

在`DefaultMQPushConsumerImpl`的start方法中，对消息监听器类型进行了判断，如果类型是`MessageListenerOrderly`表示要进行顺序消费，此时使用`ConsumeMessageOrderlyService`对`ConsumeMessageService`进行实例化，然后调用它的start方法进行启动：

public class DefaultMQPushConsumerImpl implements MQConsumerInner {
        // 消息消费service
        private ConsumeMessageService consumeMessageService;
      
        public synchronized void start() throws MQClientException {
            switch (this.serviceState) {
                case CREATE_JUST:
                    // ...
                
                    // 如果是顺序消费
                    if (this.getMessageListenerInner() instanceof MessageListenerOrderly) {
                        // 设置顺序消费标记
                        this.consumeOrderly = true;
                        // 创建consumeMessageService，使用的是ConsumeMessageOrderlyService
                        this.consumeMessageService =
                            new ConsumeMessageOrderlyService(this, (MessageListenerOrderly) this.getMessageListenerInner());
                    } else if (this.getMessageListenerInner() instanceof MessageListenerConcurrently) {
                        this.consumeOrderly = false;
                        // 并发消费使用ConsumeMessageConcurrentlyService
                        this.consumeMessageService =
                            new ConsumeMessageConcurrentlyService(this, (MessageListenerConcurrently) this.getMessageListenerInner());
                    }
                    // 启动ConsumeMessageService
                    this.consumeMessageService.start();
    
                    // ...
                    break;
              // ...
            }
            // ...
        }
    }
    复制代码

### 2.1 加锁定时任务 ###

进入到`ConsumeMessageOrderlyService`的start方法中，可以看到，如果是集群模式，会启动一个`定时加锁的任务`，`周期性的对订阅的消息队列进行加锁`，具体是通过调用`RebalanceImpl`的`lockAll`方法实现的：

public class ConsumeMessageOrderlyService implemen

[format_png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/23/ba276f6eec90480db4a26599598e1763.png
[RocketMQ_16-consumer]: https://juejin.cn/post/7219137689459294268