背景： 业务使用 RocketMQ 的场景增多，但是有一些消息状态依赖的场景没有考虑顺序
正确的使用RocketMQ 的顺序消息，能记住其中的原理
使用过RocketMQ 中间件，但是对如何使用 RocketMQ 顺序消息，以及知道如何使用但是不了解它的原理的人
耗时：8分钟

1. 顺序消息业务使用场景

• 使用RocketMQ来传递带状态的订单消息
• 使用RocketMQ来同步MySQL的 binlog 日志，这部分大部分使用kafka代替了
• 其他消息之间有先后的依赖关系，后一条消息需要依赖于前一条消息的处理结果的情况

2. 如何正确使用顺序消息

2.1 消息生产端

​ 在RocketMQ 的生产端，原始的发送客户端 API (DefaultMQProducer) 提供了多种发送消息的方式，其中一种是按用户自定义选择队列进行发送，用户只需要实现MessageQueueSelector 接口即可，具体接口代码如下：

// 自定义选择队列接口
public interface MessageQueueSelector {
    MessageQueue select(final List<MessageQueue> mqs, final Message msg, final Object arg);
} 
// 同步发送方式
SendResult send(final Message msg, final MessageQueueSelector selector, final Object arg,
                    final long timeout) 
// 异步发送方式
void send(final Message msg, final MessageQueueSelector selector, final Object arg,
              final SendCallback sendCallback) 

同时RocketMQ还提供了MessageQueueSelector接口的实现类来给用户做选择

1. SelectMessageQueueByHash, ：按 参数 arg的 hashCode 与可选队列的大小求余来选择发送队列
2. SelectMessageQueueByRandom ：在可选队列中随机选择一个MessageQueue 进行发送

2.2 消息消费端

RocketMQ原始API在消费端提供两种获取消息的方式，push 方式和pull 方式：

• 1. push 方式，broker 有消息的时候推送到客户端（这里暂时这里理解吧，对消费端使用者来说，像是broker 主动推送消息来的）
• 2. pull 方式是客户端主动拉取。

2.2.1 pull 方式顺序消费

pull 方式的原始消费实例是 DefaultMQPullConsumer 它实现了MQPullConsumer 接口，拉取消息的 API 如下:

    /**
     * @param mq 从那个队列来取
     * @param subExpression 根据tag过滤消息
     * @param offset 从按个位置拉取
     * @param maxNums 最大拉取数量
     */
    PullResult pull(final MessageQueue mq, final String subExpression, final long offset,
        final int maxNums) throws MQClientException, RemotingException, MQBrokerException,
        InterruptedException
            
    // MessageQueue 对象 可以用当前接口的另外方法拿到，即
   Set<MessageQueue> fetchSubscribeMessageQueues(final String topic) throws MQClientException

​ pull 方式要先指定拉取的队列，也就是消费端要自己维护节点要拉取的队列，而一次拉取过来的消息是顺序的，这部分的顺序消费都需要业务自己处理，这里不展开讨论。

2.2.2 push 方式消费

push 方式原生的消费实例是 DefaultMQPushConsumer , 它主要实现了 MQPushConsumer 接口，接口 API 如下：

public interface MQPushConsumer extends MQConsumer {
    //  并行消费方式
    void registerMessageListener(final MessageListenerConcurrently messageListener);
      // 串行消费方式
    void registerMessageListener(final MessageListenerOrderly messageListener);
}

RocketMQ 提供两种注册监听事件，其中实现MessageListenerOrderly 的接口监听，可以保证顺序消费

RocketMQ 通过上面简单的消费 API 就实现了顺序消费，那它是如何实现的呢？接下来看RocketMQ 的源码来看顺序消费的原理

3. RocketMQ 顺序消息的原理

​ 在了解顺序消息的原理之前，我们先看下 RocketMQ 的消息模型

rocketmq消息模型.png

1. producer 向topic发送消息，同一个topic下存在多个队列，RocketMQ 的生产者在默认发送消息时轮询选取其中一个队列进行发送，会导致消息分散到两个队列上

2. broker 上的消息只有在同一个队列中消息才是顺序读取的

3. 消费组消费消息时，每一个consumer 单独消费broker 上的一个队列，一般情况下一个consumer 一个进程，不同进程消费不能保证消费的顺序性

4. 同一个进程下，通常会配置多个线程消费，多个线程消费的情况下没办法保证消费顺序

所以rocketMQ 默认情况下是不能保证消息的顺序的。

3.2. RocketMQ 顺序消费实现原理

​ 通过上面 RocketMQ 的消息模型，要实现顺序消息，为了能记住它实现的原理，我们先换个思路想想，如果是我们自己来实现这个功能，应该如何设计？

3.2.1 生产端

​ 先看消息生产者，因为broker 上同一个主题下的多个queue 互相之间是不能保证消息的顺序的，只有单个queue 上的消息才能保证顺序，那可以思考下它的实现方案如下：

1. topic 下只配置一个queue，但这样消息量多的情况下，并发量能会大大降低，而且这样不能发挥分布式系统的优点

2. 多数情况下，一个topic 下并不是所有消息都要保证顺序的，只是部分消息要保证顺序就可以了，所以只要能保证，这部分要保证顺序的消息放到同一个队列下就可以保证生产者发送的消息是有序的。实际 rocketmq 也是提供这样实现的 API

3. 生产者发送的消息体上设置消息的顺序，消息顺序的处理完全有消费端处理，但是在分布式系统下，消费端一般都有多个节点，要保证节点之间消费的消息是顺序，节点之间就要做到互相的感知，这样增加了消费端的复杂度；
   这里应该还有其他的方案，但是作为通用的中间件来说，方案 2 是他们那个背景下的最好方案

3.2.2 消费端

​ 根据生产者在RocketMQ 通用的实现方案，生产者已经把有顺序的消息放到同一个队列中，而消费端要做的事情，实际就是保证同一个队列下的消息能串行消费，队列在borker服务端，消费处理在consumer 上，他们分别是在两个进程中； 根据这个背景，可以将这个问题拆解成三部分

顺序消费问题拆解.png

1. broke 上要保证一个队列只有一个进程消费，即一个队列同一时间只有一个consumer 消费
2. broker 给consumer 的消息顺序应该保持一致，这个通过 rpc传输，序列化后消息顺序不变，所以很容易实现
3. consumer 上的队列消息要保证同一个时间只有一个线程消费

通过问题的拆分，问题变成同一个共享资源串行处理了，要解决这个问题，通常的做法都是访问资源的时候加锁，即broker 上一个队列消息在被consumer 访问的必须加锁，单个consumer 端多线程并发处理消息的时候需要加锁；这里还需要考虑broker 锁的异常情况，假如一个broke 队列上的消息被consumer 锁住了，万一consumer 崩溃了，这个锁就释放不了，所以broker 上的锁需要加上锁的过期时间。
实际上 RocketMQ 消费端也就是照着上面的思路做：

顺序消费

1. broker 中在类 RebalanceLockManager 的静态变量 mqLockTable (变量类型为 ConcurrentMap) 中存储了以消费组 为key ， 以 ConcurrentMap （以消息主题，主题下队列为key，具体信息是消费者客户端id 为和客户端上次锁定时间 为value的 LockEntity 对象）为value的消费者锁定信息，

2. broker 接受请求后执行 RebalanceLockManager 的 tryLockBatch 方法 ，tryLockBatch 执行顺序逻辑如下：

   1. 请求参数解析，解析成 要锁定的主题下队列集合和消费者ID

   2. 遍历请求锁定的队列

   3. 通过 mqLockTable 判断单个队列是否已经锁定，即调用 LockEntry 的 isLocked 方法，主要是判断 clientId 是否是当前消费者ID，如果是就更新锁定时间，并加入已经锁定队列中，如果 mqLockTable 不存在 这个消费组或者当前锁定的clientId与请求的clientId 不相等，就加入未锁定队列

   4. 判断未锁定队列是否为空，不为空，判断当前消费组是否在mqLockTable 中，不存在就创建，后启用 RebalanceLockManager 的可重入锁，遍历未锁定队列

   5. 执行第3 步

   6. 释放 RebalanceLockManager 的可重入锁，返回当前锁定的信息

3. consumer 上顺序消费 的类里面有一个定时任务，每隔20去向broke 发送它订阅的topic 的锁定请求

4. consumer 上在获取到队列的消息的时候，让消费线程池去处理，处理前必须获取到本地队列的锁。

下面是具体的RocketMQ 实现代码。部分代码省略：

RocketMQ 顺序消费客户端的源码都是在ConsumeMessageOrderlyService 上

// 顺序消费核心逻辑
public class ConsumeMessageOrderlyService implements ConsumeMessageService {
    // 消费线程池
    private final ThreadPoolExecutor consumeExecutor; 
    public ConsumeMessageOrderlyService(defaultMQPushConsumerImpl, messageListener){
// 构造函数里面主要是初始化消费线程 ，即初始化 consumeExecutor，我们初始化参数里面 消费的最小线程数和最大线程数就是在这里用的
    }
    // 启动方法
    public void start() {
        // 启动向服务器发送broke 上队列锁定的定时任务，默认每隔20 秒执行一次
    this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
        public void run() { ConsumeMessageOrderlyService.this.lockMQPeriodically(); }
        }, 1000 * 1, ProcessQueue.REBALANCE_LOCK_INTERVAL, TimeUnit.MILLISECONDS);
      }
    // 获取服务器上的消费队列锁
    public synchronized void lockMQPeriodically() {
        if (!this.stopped) {
            this.defaultMQPushConsumerImpl.getRebalanceImpl().lockAll();
        }
    }     
    public void submitConsumeRequest( msgs, processQueue, messageQueue,  dispathToConsume) {
        //提交消费任务,从服务器上拉取到消息，提交消费任务，让线程池consumeExecutor里的线程执行消费任务
        if (dispathToConsume) {
            ConsumeRequest consumeRequest = new ConsumeRequest(processQueue, messageQueue);
            this.consumeExecutor.submit(consumeRequest);
        }
    }
    // 实际的消费任务
    class ConsumeRequest implements Runnable {
        // 本地消息队列
        private final ProcessQueue processQueue;
        // 当前消费队列 的broke名称和broker下队列id 信息
        private final MessageQueue messageQueue;
        public void run(){
            // ....校验代码省略
            final Object objLock = messageQueueLock.fetchLockObject(this.messageQueue);
            synchronized (objLock)  {  
             //  ....校验代码省略
              // 从本地队列里面获取到一次消费的消息数据  consumeBatchSize 配置里面一次消费的消息数， takeMessags 方法保证本地的消息在取consumeBatchSize 消息的时候，不会写数据到本地队列
              List<MessageExt> msgs = this.processQueue.takeMessags(consumeBatchSize);
              // msgs 校验，执行小消费前的钩子方法 省略
              try {
                  this.processQueue.getLockConsume().lock();
                  // 执行实际的消费逻辑，返回消费状态， 这里消费前调用 Collections.unmodifiableList ，保证不让用户修改消费消息的顺序
                  status = messageListener.consumeMessage(Collections.unmodifiableList(msgs), context);
                } catch (Throwable e) {
                } finally {
                   this.processQueue.getLockConsume().unlock();
                }
            }
        }
    }
}

​ consumer 顺序消费，在启动的时候，会启动定时任务每隔20 秒向服务器发送锁定队列的命令，即上面的 lockMQPeriodically 方法，这个方法实际调用 broker 上的 org.apache.rocketmq.broker.processor.AdminBrokerProcessor 上lockBatchMQ 方法，即锁定broker上当前消费的队列，broke上队列的锁定信息都在 RebalanceLockManager 保存着 ，即mqLockTable 这个 ConcurrentHashMap 中，其中 group 为消费组名称，MessageQueue 保存队列信息，LockEntry记录着锁定当前队列的 具体的consumer 客户端Id 和上次同步锁的时间，我们具体看下 RebalanceLockManager 锁的维护代码：

 /***
     * 客户端请求，尝试锁定队列
     * @param group  消费组名称
     * @param mqs 尝试获取锁的队列（包括topic，queue ，broker信息）
     * @param clientId，当前想获取锁的消费者id
     * @return
     */
    public Set<MessageQueue> tryLockBatch(final String group, final Set<MessageQueue> mqs,
                                          final String clientId) {
        Set<MessageQueue> lockedMqs = new HashSet<MessageQueue>(mqs.size());
        Set<MessageQueue> notLockedMqs = new HashSet<MessageQueue>(mqs.size());

        for (MessageQueue mq : mqs) {
            // 判断当前客户端是否已经锁定了当前队列，如果锁定了，直接加入锁定的set 中
            if (this.isLocked(group, mq, clientId)) {
                lockedMqs.add(mq);
            } else {
                notLockedMqs.add(mq);
            }
        }
        // 没有锁定，尝试进行锁定
        if (!notLockedMqs.isEmpty()) {
            this.lock.lockInterruptibly();
            try {
                // 消费组不在所锁定map中，加入锁定的map中
                ConcurrentHashMap<MessageQueue, LockEntry> groupValue = this.mqLockTable.get(group);
                if (null == groupValue) {
                    groupValue = new ConcurrentHashMap<>(32);
                    this.mqLockTable.put(group, groupValue);
                }
                for (MessageQueue mq : notLockedMqs) {
                    LockEntry lockEntry = groupValue.get(mq);
                    if (null == lockEntry) {
                        lockEntry = new LockEntry();
                        lockEntry.setClientId(clientId);
                        groupValue.put(mq, lockEntry);              
                    } 
                    // 如果已经锁定，更新最新的锁定时间，isLocked 实际的逻辑时判断lockEntry 中的clientId 和传参相同，并且在判断当前时间和上次锁定时间差是否大于60 秒，两个条件满足，为锁定成功
                    if (lockEntry.isLocked(clientId)) {
                        lockEntry.setLastUpdateTimestamp(System.currentTimeMillis());
                        lockedMqs.add(mq);
                        continue;
                    }
                    String oldClientId = lockEntry.getClientId();
                    // isExpired判断当前时间和上次锁定时间是否大于60 秒，大于60 秒，表示过期，可用进行锁定，替换clientId，更新锁定时间，加入到锁定队列
                    if (lockEntry.isExpired()) {
                        lockEntry.setClientId(clientId);
                        lockEntry.setLastUpdateTimestamp(System.currentTimeMillis());
                        lockedMqs.add(mq);
                        continue;
                    }
                }
            } finally {
                this.lock.unlock();
            }
        }
        return lockedMqs;
    }

这部分代码并不复杂，有兴趣可以下载RocketMQ 的源码，在broke 包的 RebalanceLockManager 看看。

3.3 顺序消息原理总结

1. 消息生产端需要保证有顺序的消息放到topic 的同一个队列下，这部分 RocketMQ 提供了相应的API。

2. 消费的时候，使用了两把锁，一把锁住broke上要消费的队列，一把锁在consumer 要处理的本地消息上，为了在两个进程上做到锁的一致，使用了一个定时任务来保证锁的同步。

3. broker 上的锁为了防止已经获得锁的consumer异常导致同一消费组上的其他consumer 消费不了这个队列上的消息，broker上的锁加上了过期机制。

在了解了一个中间件底层原理的时候，我们往往当时看了文档，知道了它的底层原理，但是时间久了很容易忘记，而且这样记忆是没有意义的，我们应该站在开发者的角度思考下，如果这个功能，在他们那个背景下我们自己实现，应该如何实现？然后再对照下我们自己实现方案和实际的实现方案，优劣在哪里（也有可能你的方案可能和开发者的方案是一样的）？如果是现在的背景下，方案是否还有改进？下次遇到业务上相同的场景，是否也可以按照这个方案实行？只有经过了这样的思考，我们才能加深印象，才能学到源码的精髓。

4. 顺序消息注意事项

1. 实际项目中并不是所有情况都需要用到顺序消息，但这也是设计方案的时候容易忽略的一点
2. 顺序消息是生产者和消费者配合协调作用的结果，但是消费端保证顺序消费，是保证不了顺序消息的
3. 消费端并行方式消费，只设置一次拉取消息的数量为 1（即配置参数 consumeBatchSize ），是否可以实现顺序消费 ？这里实际是不能的，并发消费在消费端有多个线程同时消费，consumeBatchSize 只是一个线程一次拉取消息的数量，对顺序消费没有意义，这里大家有兴趣可以看 ConsumeMessageConcurrentlyService 的代码，并发消费的逻辑都在哪里。

RocketMQ 的学习资料，网上有很多，但是最全的信息还是在官网 下载慢可以去码云 ,它的文档都在源码包的 docs 目录下，很全！