RocketMQ消息如何实现顺序性：原理、实现与问题解析

文
章
目
录

RocketMQ作为一款高性能的消息队列，和Kafka类似，它支持基于队列（分区）的顺序消费，即同一队列内的消息能保证有序，不同队列间的消息则无序。下面，咱们就深入探讨下在RocketMQ中如何保证消息的顺序性。

一、消息发送端保障顺序性

当我们以RocketMQ生产者的身份发送顺序消息时，需要在send方法里传入一个MessageQueueSelector。这个MessageQueueSelector里有个select方法，它的作用是决定消息该被发送到哪个MessageQueue。通常，我们会用取模法来进行路由，具体代码如下：

SendResult sendResult = producer.send(
        msg,
        new MessageQueueSelector() {
            @Override
            // mqs：该Topic下所有可选的MessageQueue
            // msg：待发送的消息
            // arg：发送消息时传递的参数
            public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
                Integer id = (Integer) arg;
                //根据参数，计算出一个要接收消息的MessageQueue的下标
                int index = id % mqs.size();
                //返回这个MessageQueue
                return mqs.get(index);
            }
        },
        orderId);

这段代码里，arg参数会被转化为整数id，然后通过id对mqs（该Topic下所有可选的MessageQueue列表）的大小取模，得到一个下标，进而确定消息要发送到的MessageQueue。需要注意的是，这里建议使用同步发送的方式，这样才能保证消息按顺序发送到指定队列。

二、消息消费端保障顺序性

消息按顺序发送到消息队列后，消费者要如何按发送顺序消费呢？RocketMQ的MessageListener回调函数提供了两种消费模式：有序消费模式MessageListenerOrderly和并发消费模式MessageListenerConcurrently。若要实现顺序消费，得采用有序消费模式，示例代码如下：

consumer.registerMessageListener(
    new MessageListenerOrderly() {
        @Override
        public ConsumeOrderlyStatus consumeMessage (List<MessageExt> msgs, ConsumeOrderlyContext context) {
            System.out.printf("Receive order msg:" + new String(msgs.get(0).getBody()));
            return ConsumeOrderlyStatus.SUCCESS;
        }
});

当我们用这种方式注册一个消费者后，为确保同一队列中的有序消息能顺序消费，还得保证RocketMQ的Broker只会把消息发送到同一个消费者上，这就需要加锁了。

在ConsumeMessageOrderlyService初始化时，会启动一个定时任务，该任务会尝试向Broker为当前消费者客户端申请分布式锁。如果成功获取锁，后续消息就会只发给这个Consumer。

消息拉取过程中，消费者会一次性拉取多条消息，并将这些消息放入ProcessQueue，同时提交到消费线程池执行。那拉取后的消费过程如何保证顺序性呢？这就涉及到更多的锁机制了。

RocketMQ在消费时，需要申请MessageQueue锁，确保同一时间只有一个线程能处理队列中的消息。获取MessageQueue锁后，就可以从ProcessQueue中依次拉取一批消息处理。但在这个过程中，为避免消息重复消费，还得对ProcessQueue加锁，这部分内容在后面扩展知识中会详细展开。之后，就可以开始处理业务逻辑了。

总的来说，为保障消息顺序性，RocketMQ在消费过程中进行了三次加锁：首先锁定Broker上的MessageQueue，确保消息只投递到唯一的消费者；接着对本地的MessageQueue加锁，保证只有一个线程能处理该消息队列；最后对存储消息的ProcessQueue加锁，防止在重平衡过程中出现消息重复消费。

三、加锁机制的扩展知识：第三把锁的作用

前面提到客户端加锁过程中有三把锁，可能有人会疑惑：第三把锁如果不加，是不是也没问题呢？毕竟我们已经对MessageQueue加锁了，为什么还需要对ProcessQueue再次加锁？

这主要是考虑到重平衡的情况。当消费者集群新增消费者，发生重平衡时，原本由客户端A消费的某个队列，可能会重新分配给客户端B。这时，客户端A需要把加在Broker上的锁解掉。但在解锁过程中，要确保消息不会在消费过程中被移除。因为如果客户端A正在处理一部分消息，而位点信息还未提交，此时若客户端B立马去消费队列中的消息，就可能导致部分数据被重复消费。

那如何判断消息是否正在消费中呢？这就要通过ProcessQueue上的锁来判断。也就是说，解锁的线程也需要尝试对ProcessQueue加锁，只有加锁成功才能进行解锁操作，以此避免在解锁过程中有消息被消费。