RocketMQ顺序消费实现原理分析

RocketMQ顺序消费实现原理简要分析(基于最新的4.7.1)

顺序消费 VS 并发消费

假如生产者投递了三条消息(M1,M2,M3)到topicM

并发消费

并发消费时如下图所示

消费顺序是乱序的，可能是M1被先消费，或者M2…

因为消费者是多线程的，每个线程得到CPU调度的时序得不到保证，每个线程都有可能消费到队列中拉取到的任意消息.

顺序消费

顺序消费时如下图

针对同一批次消息的不同状态，投递到同一队列.

拉取消息之前对队列加锁，保证同一消费组的多个消费实例，只能有一个实例能够拉取到消息.

拉取消息成功后，对消息队列使用synchronized关键字同步，消息只能由一个线程消费.

源码分析

一段简短的消费端代码,摘抄自rocketmq官网

生产者代码

public class OrderedProducer {
    public static void main(String[] args) throws Exception {
        //Instantiate with a producer group name.
        MQProducer producer = new DefaultMQProducer("example_group_name");
        //Launch the instance.
        producer.start();
        String[] tags = new String[] {"TagA", "TagB", "TagC", "TagD", "TagE"};
        for (int i = 0; i < 100; i++) {
            int orderId = i % 10;
            //Create a message instance, specifying topic, tag and message body.
            Message msg = new Message("TopicTestjjj", tags[i % tags.length], "KEY" + i,
                    ("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET));
                    //发消息时使用MessageQueueSelector,保证了相同订单id的消息发送到同一MessageQueue
            SendResult sendResult = producer.send(msg, new MessageQueueSelector() {
            @Override
            public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
                Integer id = (Integer) arg;
                int index = id % mqs.size();
                return mqs.get(index);
            }
            }, orderId);

            System.out.printf("%s%n", sendResult);
        }
        //server shutdown
        producer.shutdown();
    }
}

消费者代码

public class OrderedConsumer {
    public static void main(String[] args) throws Exception {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("example_group_name");

        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);

        consumer.subscribe("TopicTest", "TagA || TagC || TagD");

        consumer.registerMessageListener(new MessageListenerOrderly() {

            AtomicLong consumeTimes = new AtomicLong(0);
            @Override
            public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs,
                                                       ConsumeOrderlyContext context) {
                context.setAutoCommit(false);
                System.out.printf(Thread.currentThread().getName() + " Receive New Messages: " + msgs + "%n");
                this.consumeTimes.incrementAndGet();
                if ((this.consumeTimes.get() % 2) == 0) {
                    return ConsumeOrderlyStatus.SUCCESS;
                } else if ((this.consumeTimes.get() % 3) == 0) {
                    return ConsumeOrderlyStatus.ROLLBACK;
                } else if ((this.consumeTimes.get() % 4) == 0) {
                    return ConsumeOrderlyStatus.COMMIT;
                } else if ((this.consumeTimes.get() % 5) == 0) {
                    context.setSuspendCurrentQueueTimeMillis(3000);
                    return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
                }
                return ConsumeOrderlyStatus.SUCCESS;

            }
        });

        consumer.start();

        System.out.printf("Consumer Started.%n");
    }
}

拉消息流程

上述代码在给consumer注册消息监听器时，使用了MessageListenerOrderly(关键)

consumer#start()启动逻辑

public void start() throws MQClientException {
       setConsumerGroup(NamespaceUtil.wrapNamespace(this.getNamespace(), this.consumerGroup));
       //
       this.defaultMQPushConsumerImpl.start();
       if (null != traceDispatcher) {
           try {
               traceDispatcher.start(this.getNamesrvAddr(), this.getAccessChannel());
           } catch (MQClientException e) {
               log.warn("trace dispatcher start failed ", e);
           }
       }
   }

调用org.apache.rocketmq.client.impl.consumer.DefaultMQPushConsumerImpl#start

public synchronized void start() throws MQClientException {
        switch (this.serviceState) {
            //....
            //根据服务状态，做不同的逻辑处理
            //初始化 mqClientFactory
            //初始化RebalanceImpl(设置消费模型，设置消费组，设置消息队列分配策略)
            //根据消息监听器的实例类型确定是否顺序消费，然后初始化ConsumeMessageService
        }

        this.updateTopicSubscribeInfoWhenSubscriptionChanged();
        this.mQClientFactory.checkClientInBroker();
        this.mQClientFactory.sendHeartbeatToAllBrokerWithLock();
        //立即拉取消息
        this.mQClientFactory.rebalanceImmediately();
    }

上述RebalanceImpl的实现类为(RebalancePushImpl)

rebalanceImmediately

public void rebalanceImmediately() {
       this.rebalanceService.wakeup();
   }

唤醒RebalanceService

RebalanceService#run

// 默认等待时间20秒
private static long waitInterval =
        Long.parseLong(System.getProperty(
            "rocketmq.client.rebalance.waitInterval", "20000"));

 @Override
    public void run() {
        log.info(this.getServiceName() + " service started");

        while (!this.isStopped()) {
            //这里等待间隔为20s秒，内部使用了重写的ContdownLatch2

            this.waitForRunning(waitInterval);
            this.mqClientFactory.doRebalance();
        }

        log.info(this.getServiceName() + " service end");
    }

mqClientFactory.doRebalance()

public void doRebalance() {
    // key为consumerGroup,value为MQConsumer实例
     for (Map.Entry<String, MQConsumerInner> entry : this.consumerTable.entrySet()) {
         MQConsumerInner impl = entry.getValue();
         if (impl != null) {
             try {
                 impl.doRebalance();
             } catch (Throwable e) {
                 log.error("doRebalance exception", e);
             }
         }
     }
 }

MQConsumerInner#doRebalance

此处MQConsumerInner的实现类为DefaultMqPushConsumerImpl

@Override
  public void doRebalance() {
      if (!this.pause) {
          this.rebalanceImpl.doRebalance(this.isConsumeOrderly());
      }
  }

rebalanceImpl.doRebalance

此处的rebalanceImpl实现类为RebalancePushImpl

public void doRebalance(final boolean isOrder) {
    //key为topic,value为相关订阅配置信息
       Map<String, SubscriptionData> subTable = this.getSubscriptionInner();
       if (subTable != null) {
           //遍历每一个topic，拉取消息
           for (final Map.Entry<String, SubscriptionData> entry : subTable.entrySet()) {
               final String topic = entry.getKey();
               try {
                   this.rebalanceByTopic(topic, isOrder);
               } catch (Throwable e) {
                   if (!topic.startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX)) {
                       log.warn("rebalanceByTopic Exception", e);
                   }
               }
           }
       }

       this.truncateMessageQueueNotMyTopic();
   }

rebalanceImpl.rebalanceByTopic

private void rebalanceByTopic(final String topic, final boolean isOrder) {
        switch (messageModel) {
            //广播模式下拉取对应topic所有队列的消息
            //也就是说同一消费组下的每个consumer都可以消费同一topic下的每条消息
            case BROADCASTING: {
                Set<MessageQueue> mqSet = this.topicSubscribeInfoTable.get(topic);
                if (mqSet != null) {
                    boolean changed = this.updateProcessQueueTableInRebalance(topic, mqSet, isOrder);
                    if (changed) {
                        this.messageQueueChanged(topic, mqSet, mqSet);
                        log.info("messageQueueChanged {} {} {} {}",
                            consumerGroup,
                            topic,
                            mqSet,
                            mqSet);
                    }
                } else {
                    log.warn("doRebalance, {}, but the topic[{}] not exist.", consumerGroup, topic);
                }
                break;
            }
            //集群模式下根据队列分配策略，拉取分配到的队列的消息
            case CLUSTERING: {
                Set<MessageQueue> mqSet = this.topicSubscribeInfoTable.get(topic);
                List<String> cidAll = this.mQClientFactory.findConsumerIdList(topic, consumerGroup);
                if (null == mqSet) {
                    if (!topic.startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX)) {
                        log.warn("doRebalance, {}, but the topic[{}] not exist.", consumerGroup, topic);
                    }
                }

                if (null == cidAll) {
                    log.warn("doRebalance, {} {}, get consumer id list failed", consumerGroup, topic);
                }

                if (mqSet != null && cidAll != null) {
                    List<MessageQueue> mqAll = new ArrayList<MessageQueue>();
                    mqAll.addAll(mqSet);

                    Collections.sort(mqAll);
                    Collections.sort(cidAll);

                    AllocateMessageQueueStrategy strategy = this.allocateMessageQueueStrategy;

                    List<MessageQueue> allocateResult = null;
                    try {
                        allocateResult = strategy.allocate(
                            this.consumerGroup,
                            this.mQClientFactory.getClientId(),
                            mqAll,
                            cidAll);
                    } catch (Throwable e) {
                        log.error("AllocateMessageQueueStrategy.allocate Exception. allocateMessageQueueStrategyName={}", strategy.getName(),
                            e);
                        return;
                    }

                    Set<MessageQueue> allocateResultSet = new HashSet<MessageQueue>();
                    if (allocateResult != null) {
                        allocateResultSet.addAll(allocateResult);
                    }

                    boolean changed = this.updateProcessQueueTableInRebalance(topic, allocateResultSet, isOrder);
                    if (changed) {
                        log.info(
                            "rebalanced result changed. allocateMessageQueueStrategyName={}, group={}, topic={}, clientId={}, mqAllSize={}, cidAllSize={}, rebalanceResultSize={}, rebalanceResultSet={}",
                            strategy.getName(), consumerGroup, topic, this.mQClientFactory.getClientId(), mqSet.size(), cidAll.size(),
                            allocateResultSet.size(), allocateResultSet);
                        this.messageQueueChanged(topic, mqSet, allocateResultSet);
                    }
                }
                break;
            }
            default:
                break;
        }
    }

rebalanceImpl#updateProcessQueueTableInRebalance

/**
 *
 * @param topic 消息主题
 * @param mqSet 消息队列集合
 * @param isOrder 是否顺序消费
 */
private boolean updateProcessQueueTableInRebalance(final String topic, final Set<MessageQueue> mqSet,
        final boolean isOrder) {
        boolean changed = false;

        //每个MessageQueue对应一个ProcessQueue(mq的快照)
        Iterator<Entry<MessageQueue, ProcessQueue>> it = this.processQueueTable.entrySet().iterator();
        while (it.hasNext()) {
            Entry<MessageQueue, ProcessQueue> next = it.next();
            MessageQueue mq = next.getKey();
            ProcessQueue pq = next.getValue();

            if (mq.getTopic().equals(topic)) {
                if (!mqSet.contains(mq)) {
                    pq.setDropped(true);
                    if (this.removeUnnecessaryMessageQueue(mq, pq)) {
                        it.remove();
                        changed = true;
                        log.info("doRebalance, {}, remove unnecessary mq, {}", consumerGroup, mq);
                    }
                    //检查距离上次拉取消息的时间是否大于拉取的最大空闲时间（默认120s)
                } else if (pq.isPullExpired()) {
                    switch (this.consumeType()) {
                        case CONSUME_ACTIVELY:
                            break;
                        case CONSUME_PASSIVELY:
                            pq.setDropped(true);
                            if (this.removeUnnecessaryMessageQueue(mq, pq)) {
                                it.remove();
                                changed = true;
                                log.error("[BUG]doRebalance, {}, remove unnecessary mq, {}, because pull is pause, so try to fixed it",
                                    consumerGroup, mq);
                            }
                            break;
                        default:
                            break;
                    }
                }
            }
        }
        //拉消息的requetTask
        List<PullRequest> pullRequestList = new ArrayList<PullRequest>();
        for (MessageQueue mq : mqSet) {
            if (!this.processQueueTable.containsKey(mq)) {
                //客户端顺序消费，则锁定对应的mq
                if (isOrder && !this.lock(mq)) {
                    //加锁失败的话，默认就忽略该mq
                    log.warn("doRebalance, {}, add a new mq failed, {}, because lock failed", consumerGroup, mq);
                    continue;
                }

                this.removeDirtyOffset(mq);
                ProcessQueue pq = new ProcessQueue();
                long nextOffset = this.computePullFromWhere(mq);
                if (nextOffset >= 0) {
                    ProcessQueue pre = this.processQueueTable.putIfAbsent(mq, pq);
                    if (pre != null) {
                        log.info("doRebalance, {}, mq already exists, {}", consumerGroup, mq);
                    } else {
                        //创建PullRequest
                        log.info("doRebalance, {}, add a new mq, {}", consumerGroup, mq);
                        PullRequest pullRequest = new PullRequest();
                        pullRequest.setConsumerGroup(consumerGroup);
                        pullRequest.setNextOffset(nextOffset);
                        pullRequest.setMessageQueue(mq);
                        pullRequest.setProcessQueue(pq);
                        pullRequestList.add(pullRequest);
                        changed = true;
                    }
                } else {
                    log.warn("doRebalance, {}, add new mq failed, {}", consumerGroup, mq);
                }
            }
        }
        //把PullRequest提交到PullMessageServcie中的pullRequestQueue
        this.dispatchPullRequest(pullRequestList);
        return changed;
    }

实际拉消息的逻辑DefaultMQPushConsumerImpl#pullMessage

public void pullMessage(final PullRequest pullRequest) {
    final ProcessQueue processQueue = pullRequest.getProcessQueue();
       if (processQueue.isDropped()) {
           log.info("the pull request[{}] is dropped.", pullRequest.toString());
           return;
       }
       //设置最新的拉取消息时间
       processQueue.setLastPullTimestamp(System.currentTimeMillis());
       //...省略
       //拉取消息后的回调
       PullCallback pullCallback = new PullCallback() {
           @Override
           public void onSuccess(PullResult pullResult) {
               if (pullResult != null) {
                   pullResult = DefaultMQPushConsumerImpl.this.pullAPIWrapper.processPullResult(pullRequest.getMessageQueue(), pullResult,
                       subscriptionData);

                   switch (pullResult.getPullStatus()) {
                       case FOUND:
                           long prevRequestOffset = pullRequest.getNextOffset();
                           pullRequest.setNextOffset(pullResult.getNextBeginOffset());
                           long pullRT = System.currentTimeMillis() - beginTimestamp;
                           DefaultMQPushConsumerImpl.this.getConsumerStatsManager().incPullRT(pullRequest.getConsumerGroup(),
                               pullRequest.getMessageQueue().getTopic(), pullRT);

                           long firstMsgOffset = Long.MAX_VALUE;
                           //拉取到的消息为空
                           if (pullResult.getMsgFoundList() == null || pullResult.getMsgFoundList().isEmpty()) {
                               //把PullRequest再次提交到任务队列
                               DefaultMQPushConsumerImpl.this.executePullRequestImmediately(pullRequest);
                           } else {
                               firstMsgOffset = pullResult.getMsgFoundList().get(0).getQueueOffset();

                               DefaultMQPushConsumerImpl.this.getConsumerStatsManager().incPullTPS(pullRequest.getConsumerGroup(),
                                   pullRequest.getMessageQueue().getTopic(), pullResult.getMsgFoundList().size());
                               //把拉取到消息放入到processQueue中
                               boolean dispatchToConsume = processQueue.putMessage(pullResult.getMsgFoundList());
                              //提交消费消费请求到消费线程池
                              //这里的consumeMessageService实际类型为ConsumeMessageOrderlyService
                               DefaultMQPushConsumerImpl.this.consumeMessageService.submitConsumeRequest(
                                   pullResult.getMsgFoundList(),
                                   processQueue,
                                   pullRequest.getMessageQueue(),
                                   dispatchToConsume);

                               if (DefaultMQPushConsumerImpl.this.defaultMQPushConsumer.getPullInterval() > 0) {
                                   DefaultMQPushConsumerImpl.this.executePullRequestLater(pullRequest,
                                       DefaultMQPushConsumerImpl.this.defaultMQPushConsumer.getPullInterval());
                               } else {
                                   DefaultMQPushConsumerImpl.this.executePullRequestImmediately(pullRequest);
                               }
                           }

                           if (pullResult.getNextBeginOffset() < prevRequestOffset
                               || firstMsgOffset < prevRequestOffset) {
                               log.warn(
                                   "[BUG] pull message result maybe data wrong, nextBeginOffset: {} firstMsgOffset: {} prevRequestOffset: {}",
                                   pullResult.getNextBeginOffset(),
                                   firstMsgOffset,
                                   prevRequestOffset);
                           }
                           break;
                           //...其他省略
                   }
               }
           }
}
 try {
       //实际拉取消息的请求
           this.pullAPIWrapper.pullKernelImpl(
               pullRequest.getMessageQueue(),
               subExpression,
               subscriptionData.getExpressionType(),
               subscriptionData.getSubVersion(),
               pullRequest.getNextOffset(),
               this.defaultMQPushConsumer.getPullBatchSize(),
               sysFlag,
               commitOffsetValue,
               BROKER_SUSPEND_MAX_TIME_MILLIS,
               CONSUMER_TIMEOUT_MILLIS_WHEN_SUSPEND,
               CommunicationMode.ASYNC,
               pullCallback
           );
       } catch (Exception e) {
           log.error("pullKernelImpl exception", e);
           this.executePullRequestLater(pullRequest, pullTimeDelayMillsWhenException);
       }
}

消息消费

定位到ConsumeMessageOrderlyService

public class ConsumeMessageConcurrentlyService implements ConsumeMessageService {
    private static final InternalLogger log = ClientLogger.getLog();
    private final DefaultMQPushConsumerImpl defaultMQPushConsumerImpl;
    private final DefaultMQPushConsumer defaultMQPushConsumer;
    private final MessageListenerConcurrently messageListener;
   
    private final BlockingQueue<Runnable> consumeRequestQueue;
     //消息消费线程池，默认20个线程
    private final ThreadPoolExecutor consumeExecutor;
    private final String consumerGroup;
    //用于消息消费阻塞后的，重新提交消费请求的任务调度线程池，默认单线程
    private final ScheduledExecutorService scheduledExecutorService;
    //清理因消息消费线程阻塞，导致消息积压的线程池,默认单线程，消息默认最大积压时间（15min）
    private final ScheduledExecutorService cleanExpireMsgExecutors;
    //...省略
    public void start() {
        //如果消息消费模型是集群
        if (MessageModel.CLUSTERING.equals(ConsumeMessageOrderlyService.this.defaultMQPushConsumerImpl.messageModel())) {
            //每隔20秒会执行一次锁定消息队列的请求
            this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
                @Override
                public void run() {
                    ConsumeMessageOrderlyService.this.lockMQPeriodically();
                }
            }, 1000 * 1, ProcessQueue.REBALANCE_LOCK_INTERVAL, TimeUnit.MILLISECONDS);
        }
    }   
}

ConsumeRequest

该类实现了Runnable接口,为实际的消息消费处理入口

run方法逻辑

public void run() {
            if (this.processQueue.isDropped()) {
                log.warn("run, the message queue not be able to consume, because it's dropped. {}", this.messageQueue);
                return;
            }
            //获取对应队列的objLock
            final Object objLock = messageQueueLock.fetchLockObject(this.messageQueue);
            //保证同一队列下只能由某个线程能够处理该队列下的消息
            synchronized (objLock) {
                if (MessageModel.BROADCASTING.equals(ConsumeMessageOrderlyService.this.defaultMQPushConsumerImpl.messageModel())
                    || (this.processQueue.isLocked() && !this.processQueue.isLockExpired())) {
                    final long beginTime = System.currentTimeMillis();
                    for (boolean continueConsume = true; continueConsume; ) {
                        if (this.processQueue.isDropped()) {
                            log.warn("the message queue not be able to consume, because it's dropped. {}", this.messageQueue);
                            break;
                        }

                        if (MessageModel.CLUSTERING.equals(ConsumeMessageOrderlyService.this.defaultMQPushConsumerImpl.messageModel())
                            && !this.processQueue.isLocked()) {
                            log.warn("the message queue not locked, so consume later, {}", this.messageQueue);
                            ConsumeMessageOrderlyService.this.tryLockLaterAndReconsume(this.messageQueue, this.processQueue, 10);
                            break;
                        }

                        if (MessageModel.CLUSTERING.equals(ConsumeMessageOrderlyService.this.defaultMQPushConsumerImpl.messageModel())
                            && this.processQueue.isLockExpired()) {
                            log.warn("the message queue lock expired, so consume later, {}", this.messageQueue);
                            ConsumeMessageOrderlyService.this.tryLockLaterAndReconsume(this.messageQueue, this.processQueue, 10);
                            break;
                        }

                        long interval = System.currentTimeMillis() - beginTime;
                        if (interval > MAX_TIME_CONSUME_CONTINUOUSLY) {
                            ConsumeMessageOrderlyService.this.submitConsumeRequestLater(processQueue, messageQueue, 10);
                            break;
                        }

                        final int consumeBatchSize =
                            ConsumeMessageOrderlyService.this.defaultMQPushConsumer.getConsumeMessageBatchMaxSize();

                        List<MessageExt> msgs = this.processQueue.takeMessages(consumeBatchSize);
                        defaultMQPushConsumerImpl.resetRetryAndNamespace(msgs, defaultMQPushConsumer.getConsumerGroup());
                        if (!msgs.isEmpty()) {
                            final ConsumeOrderlyContext context = new ConsumeOrderlyContext(this.messageQueue);

                            ConsumeOrderlyStatus status = null;

                            ConsumeMessageContext consumeMessageContext = null;
                            if (ConsumeMessageOrderlyService.this.defaultMQPushConsumerImpl.hasHook()) {
                                consumeMessageContext = new ConsumeMessageContext();
                                consumeMessageContext
                                    .setConsumerGroup(ConsumeMessageOrderlyService.this.defaultMQPushConsumer.getConsumerGroup());
                                consumeMessageContext.setNamespace(defaultMQPushConsumer.getNamespace());
                                consumeMessageContext.setMq(messageQueue);
                                consumeMessageContext.setMsgList(msgs);
                                consumeMessageContext.setSuccess(false);
                                // init the consume context type
                                consumeMessageContext.setProps(new HashMap<String, String>());
                                ConsumeMessageOrderlyService.this.defaultMQPushConsumerImpl.executeHookBefore(consumeMessageContext);
                            }

                            long beginTimestamp = System.currentTimeMillis();
                            ConsumeReturnType returnType = ConsumeReturnType.SUCCESS;
                            boolean hasException = false;
                            try {
                                //这里还需要加锁，因为消费过程中，有可能在创建拉消息的请求时该ProceessQueue对应的MessageQueue被移除掉,或者距离上次拉消息的时间超过最大空闲时间（120）
                                this.processQueue.getLockConsume().lock();
                                //上述情况发生后，processQueue被标记为dropped
                                if (this.processQueue.isDropped()) {
                                    //消息不作处理
                                    log.warn("consumeMessage, the message queue not be able to consume, because it's dropped. {}",
                                        this.messageQueue);
                                    break;
                                }
                                //调用业务代码设置的的消息监听器的回调方法
                                status = messageListener.consumeMessage(Collections.unmodifiableList(msgs), context);
                            } catch (Throwable e) {
                                log.warn("consumeMessage exception: {} Group: {} Msgs: {} MQ: {}",
                                    RemotingHelper.exceptionSimpleDesc(e),
                                    ConsumeMessageOrderlyService.this.consumerGroup,
                                    msgs,
                                    messageQueue);
                                hasException = true;
                            } finally {
                                this.processQueue.getLockConsume().unlock();
                            }

                            if (null == status
                                || ConsumeOrderlyStatus.ROLLBACK == status
                                || ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT == status) {
                                log.warn("consumeMessage Orderly return not OK, Group: {} Msgs: {} MQ: {}",
                                    ConsumeMessageOrderlyService.this.consumerGroup,
                                    msgs,
                                    messageQueue);
                            }

                            long consumeRT = System.currentTimeMillis() - beginTimestamp;
                            if (null == status) {
                                if (hasException) {
                                    returnType = ConsumeReturnType.EXCEPTION;
                                } else {
                                    returnType = ConsumeReturnType.RETURNNULL;
                                }
                            } else if (consumeRT >= defaultMQPushConsumer.getConsumeTimeout() * 60 * 1000) {
                                returnType = ConsumeReturnType.TIME_OUT;
                            } else if (ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT == status) {
                                returnType = ConsumeReturnType.FAILED;
                            } else if (ConsumeOrderlyStatus.SUCCESS == status) {
                                returnType = ConsumeReturnType.SUCCESS;
                            }

                            if (ConsumeMessageOrderlyService.this.defaultMQPushConsumerImpl.hasHook()) {
                                consumeMessageContext.getProps().put(MixAll.CONSUME_CONTEXT_TYPE, returnType.name());
                            }

                            if (null == status) {
                                status = ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
                            }

                            if (ConsumeMessageOrderlyService.this.defaultMQPushConsumerImpl.hasHook()) {
                                consumeMessageContext.setStatus(status.toString());
                                consumeMessageContext
                                    .setSuccess(ConsumeOrderlyStatus.SUCCESS == status || ConsumeOrderlyStatus.COMMIT == status);
                                ConsumeMessageOrderlyService.this.defaultMQPushConsumerImpl.executeHookAfter(consumeMessageContext);
                            }

                            ConsumeMessageOrderlyService.this.getConsumerStatsManager()
                                .incConsumeRT(ConsumeMessageOrderlyService.this.consumerGroup, messageQueue.getTopic(), consumeRT);

                            continueConsume = ConsumeMessageOrderlyService.this.processConsumeResult(msgs, status, context, this);
                        } else {
                            continueConsume = false;
                        }
                    }
                } else {
                    if (this.processQueue.isDropped()) {
                        log.warn("the message queue not be able to consume, because it's dropped. {}", this.messageQueue);
                        return;
                    }

                    ConsumeMessageOrderlyService.this.tryLockLaterAndReconsume(this.messageQueue, this.processQueue, 100);
                }
            }
        }

    }

总结

大致流程图

值得注意的是，Consumer在对broker端的消息队列加锁时，在Consumer实际发出拉取消息的请求后，broker此时收到请求后，并未对请求的消息队列做是否加锁验证.

所以，总体而言，rocketmq为了保证消息的局部消费有序，做了以下工作

1. 首先Producer得保证同一批次的不同状态的消息得投递到相同的MessageQueue

2. Consumer在拉取消息之前对要拉取的MessageQueue加锁，成功后才拉取消息

3. Consumer在拉取消息成功后，保证同一MessageQueue的消息只能由线程池中的某个线程处理.

参考

RocketMq documentaion

RocketMQ-顺序消息Demo及实现原理分析