RocketMQ有序消费模式源码解析及关键点总结

﹏ヽ暗。殇╰゛Y 2022-05-28 00:43 227阅读 0赞

有序消费模式和并发消费模式有一些关键地方不同，我这里只讲相对于并发模式下的不同点，以及对这些不同点做一些总结，说明为什么如此，所以看这篇博客之前希望你对并发消费模式的原理有一定的了解。

有序消费一般需要和有序发送一起配合使用，有序发送通过MessageQueueSelector来指定消息发送到的MessageQueue，尽量让有序消息发送到同一个MessageQueue，这是有序消费的基础，很简单，这里就不讲了，可自行百度。

**1.** 和并发模式一样，每一个订阅的MessageQueue的第一个消息拉取请求是由负载均衡后发起的，在负载均衡后，订阅了新的MessageQueue，此时不同于并发模式，有序模式需要向Broker请求MessageQueue的分布式锁，成功了才会发起第一个消息拉取请求。

public abstract class RebalanceImpl { 
    
        private boolean updateProcessQueueTableInRebalance(final String topic, final Set<MessageQueue> mqSet, final boolean isOrder) {
            ......
            // 增加 不在processQueueTable && 存在于mqSet 里的消息队列。
            List<PullRequest> pullRequestList = new ArrayList<>();
            for (MessageQueue mq : mqSet) {
                if (!this.processQueueTable.containsKey(mq)) {
                    if (isOrder && !this.lock(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);  //从Broker读取MessageQueue的消费OffSet
                    if (nextOffset >= 0) {
                        ProcessQueue pre = this.processQueueTable.putIfAbsent(mq, pq);
                        if (pre != null) {
                            log.info("doRebalance, {}, mq already exists, {}", consumerGroup, mq);
                        } else {
                            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);
                    }
                }
            }
            // 发起消息拉取请求
            this.dispatchPullRequest(pullRequestList);
    
            return changed;
        }
    
        /** * 向Broker请求获得指定消息队列的分布式锁 * * @param mq 队列 * @return 是否成功 */
        public boolean lock(final MessageQueue mq) {
            FindBrokerResult findBrokerResult = this.mQClientFactory.findBrokerAddressInSubscribe(mq.getBrokerName(), MixAll.MASTER_ID, true);
            if (findBrokerResult != null) {
                LockBatchRequestBody requestBody = new LockBatchRequestBody();
                requestBody.setConsumerGroup(this.consumerGroup);
                requestBody.setClientId(this.mQClientFactory.getClientId());
                requestBody.getMqSet().add(mq);
    
                try {
                    // 请求Broker获得指定消息队列的分布式锁
                    Set<MessageQueue> lockedMq = this.mQClientFactory.getMQClientAPIImpl().lockBatchMQ(findBrokerResult.getBrokerAddr(), requestBody, 1000);
    
                    // 设置消息处理队列锁定成功。锁定消息队列成功，可能本地没有消息处理队列，设置锁定成功会在lockAll()方法。
                    for (MessageQueue mmqq : lockedMq) {
                        ProcessQueue processQueue = this.processQueueTable.get(mmqq);
                        if (processQueue != null) {
                            processQueue.setLocked(true);
                            processQueue.setLastLockTimestamp(System.currentTimeMillis());
                        }
                    }
    
                    boolean lockOK = lockedMq.contains(mq);
                    log.info("the message queue lock {}, {} {}",
                        lockOK ? "OK" : "Failed",
                        this.consumerGroup,
                        mq);
                    return lockOK;
                } catch (Exception e) {
                    log.error("lockBatchMQ exception, " + mq, e);
                }
            }
    
            return false;
        }
    
    }

**2.** 一个MessageQueue同一时间只能被一个消费者锁定，消费者每隔20S向Broker更新最后锁定时间。在请求分布式锁时，如果一个MessageQueue的最后更新时间对比当前超过了60S，则锁过期，此时MessageQueue能被其他消费者锁定。

public class ConsumeMessageOrderlyService implements ConsumeMessageService { 
    
        public void start() {
            if (MessageModel.CLUSTERING.equals(ConsumeMessageOrderlyService.this.defaultMQPushConsumerImpl.messageModel())) {
                this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
                    @Override
                    public void run() {
                        ConsumeMessageOrderlyService.this.lockMQPeriodically();
                    }
                }, 1000 * 1, ProcessQueue.REBALANCE_LOCK_INTERVAL, TimeUnit.MILLISECONDS);
            }
        }
    
        /** * 默认每20S锁定当前持有的所有MessageQueue */
        public synchronized void lockMQPeriodically() {
            if (!this.stopped) {
                this.defaultMQPushConsumerImpl.getRebalanceImpl().lockAll();
            }
        }
    
    }

**3.** 负载均衡后，在移除不属于自己的MessageQueue时，如果ConsumeMessageOrderlyService正在执行消费方法，消费上批次的消息，ConsumeMessageOrderlyService会给ProcessQueue加上锁，这次负载均衡就得不到此ProcessQueue的锁，也就不会释放相应的MessageQueue，所以只能等待相应MessageQueue在Broker的锁定时间过期后再被其他消费者锁定

public abstract class RebalanceImpl { 
    
        /** * 当负载均衡时,依据从Namesrv获取到的最新的{@link TopicRouteData}更新{@link #processQueueTable}里的 : MessageQueue => ProcessQueue * 若Broker或者Consumer有变化,新增了或者宕机了: * - 移除 在processQueueTable && 不存在于 mqSet 里的消息队列 * - 增加 不在processQueueTable && 存在于 mqSet 里的消息队列 * - 发起对新的MessageQueue的消费请求 * * @param topic Topic * @param mqSet 负载均衡结果后最新的消息队列数组 * @param isOrder 是否顺序 * @return 是否变更 */
        private boolean updateProcessQueueTableInRebalance(final String topic, final Set<MessageQueue> mqSet, final boolean isOrder) {
            boolean changed = false;
    
            // 移除 在processQueueTable && 不存在于 mqSet 里的消息队列
            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);   //设置此队列被丢弃,不会再PullMessage
                        if (this.removeUnnecessaryMessageQueue(mq, pq)) {      //移除不在属于自己的MessageQueue
                            it.remove();
                            changed = true;
                            log.info("doRebalance, {}, remove unnecessary mq, {}", consumerGroup, mq);
                        }
                    } else if (pq.isPullExpired()) { // 队列拉取超时，进行清理
                        switch (this.consumeType()) {
                            case CONSUME_ACTIVELY:
                                break;
                            case CONSUME_PASSIVELY:  //Push形式
                                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;
                        }
                    }
                }
            }
            ......
        }
    
    }
    
    
    public class RebalancePushImpl extends RebalanceImpl { 
    
        /** * 移除不需要的队列相关的信息 * 1. 持久化消费进度，并移除之 * 2. 顺序消费&集群模式，解锁对该队列的锁定 * * @param mq 消息队列 * @param pq 消息处理队列 * @return 是否移除成功 */
        @Override
        public boolean removeUnnecessaryMessageQueue(MessageQueue mq, ProcessQueue pq) {
            // 同步队列的消费进度，并移除之。
            this.defaultMQPushConsumerImpl.getOffsetStore().persist(mq);
            this.defaultMQPushConsumerImpl.getOffsetStore().removeOffset(mq);
            // 集群模式下，顺序消费移除时，解锁对队列的锁定
            if (this.defaultMQPushConsumerImpl.isConsumeOrderly() && MessageModel.CLUSTERING.equals(this.defaultMQPushConsumerImpl.messageModel())) {
                try {
                    // 有序消费模式下,ConsumeMessageOrderlyService在执行 MessageListenerOrderly#consumeMessage() 时会锁定ProcessQueue,
                    // 执行此方法完后才会释放ProcessQueue的Lock,见ConsumeMessageOrderlyService#464 行
                    // 锁定不成功说明在长达1S的时间内一直都在执行 MessageListenerOrderly#consumeMessage 方法
                    // 也就是说 consumeMessage() 方法执行时间超过1S,每条消息的消费速度以及方法参数List<MessageExt>的大小可能存在问题
                    // 此时只能等待相应MessageQueue在Broker的锁定时间过期后再被其他消费者锁定
                    if (pq.getLockConsume().tryLock(1000, TimeUnit.MILLISECONDS)) {
                        try {
                            return this.unlockDelay(mq, pq);
                        } finally {
                            pq.getLockConsume().unlock();
                        }
                    } else {
                        log.warn("[WRONG]mq is consuming, so can not unlock it, {}. maybe hanged for a while, {}", mq, pq.getTryUnlockTimes());
                        pq.incTryUnlockTimes();
                    }
                } catch (Exception e) {
                    log.error("removeUnnecessaryMessageQueue Exception", e);
                }
    
                return false;
            }
            return true;
        }
    
        /** * {@link ProcessQueue#hasTempMessage()}也就是{@link ProcessQueue#msgTreeMap}里有消息 * 因为已经设置了{@link ProcessQueue#dropped = true},所以不会再消费msgTreeMap里的剩余消息 * 但需要等待之前已消费消息的结果处理完,提交最后的消费进度等,最后才释放MessageQueue的分布式锁,延迟20S释放 * * @param mq 消息队列 * @param pq 消息处理队列 * @return 是否解锁成功 */
        private boolean unlockDelay(final MessageQueue mq, final ProcessQueue pq) {
            if (pq.hasTempMessage()) {
                log.info("[{}]unlockDelay, begin {} ", mq.hashCode(), mq);
                this.defaultMQPushConsumerImpl.getmQClientFactory().getScheduledExecutorService().schedule(new Runnable() {
                    @Override
                    public void run() {
                        log.info("[{}]unlockDelay, execute at once {}", mq.hashCode(), mq);
                        RebalancePushImpl.this.unlock(mq, true);
                    }
                }, UNLOCK_DELAY_TIME_MILLS, TimeUnit.MILLISECONDS);
            } else {
                this.unlock(mq, true);
            }
            return true;
        }
    
    }

有序模式下，负载均衡移除不在属于自己的MessageQueue时，有时候可能不会马上释放相应的分布式锁，极端情况下要等锁时间失效，很多情况下是延迟20S释放锁，当然如果消费压力不大的时候也会马上释放锁。所以有序消费模式下，新增加或者减少了消费者引起的负载均衡，很可能不会第一时间就能获取到消息。

这种设置我觉得是为了保证消息的严格有序，**尽量使一个MessageQueue在同一时间只有一个消费者在消费，否则可能会出现消息乱序问题**。 比如上一个消费者拉取了序号为：5，6，7，8，9的消息，但在消费到8的时候触发了负载均衡，因为此时会这是ProcessQueue的droped = true ，所以\#9消费不会再消费。但是5，6，7，8的消费进度还没有提交给Broker，下一个消费者可能有重新拉取了5，6，7，8，9号消息，从5开始消费。一个消费者消费\#8消息，一个消费\#5消息，消息乱序了。

**4.** 并发模式和有序模式在拉取到消息后的处理也不同。在并发模式下，拉取到消息后，将消息合并入ProcessQueue的TreeMap，然后立即提交消费请求到消费线程池；而有序模式不同，有序模式在将拉取到的消息合并入ProcessQueue的TreeMap，此步骤会返回一个boolean变量dispathToConsume，此变量揭示TreeMap中是否还保留上次拉取的消费未消费完。并发模式忽略此值，直接提交消费请求；而有序模式只有当此值为true，也就是上一批次的消息消费完了才会提交消费请求。

public class DefaultMQPushConsumerImpl implements MQConsumerInner { 
    
        /** * 拉取消息 * * @param pullRequest 拉取消息请求 */
        public void pullMessage(final PullRequest pullRequest) {
            ......
            // 提交拉取到的消息到ProcessQueue的TreeMap中
            // 返回 true : 上一批次的消息已经消费完了
            // 返回 false: 上一批次的消息还没消费完
            boolean dispathToConsume = processQueue.putMessage(pullResult.getMsgFoundList());
            ......
    
        }
    
    }
    
    
    /** * Queue consumption snapshot * MessageQueue消费量快照 */
    public class ProcessQueue { 
    
        /** * 添加消息，并返回是否提交给消费者 * 返回true，表示上一批次的Message已经消费完了, * ${@link #consuming}的值在${@link #takeMessags(int)}里修改,也就是有序消费时会重置${@link #consuming = false} * * @param msgs 消息 * @return 是否提交给消费者 */
        public boolean putMessage(final List<MessageExt> msgs) {
            boolean dispatchToConsume = false;
            try {
                this.lockTreeMap.writeLock().lockInterruptibly();
                try {
                    // 添加消息
                    int validMsgCnt = 0;
                    for (MessageExt msg : msgs) {
                        MessageExt old = msgTreeMap.put(msg.getQueueOffset(), msg);
                        if (null == old) {
                            validMsgCnt++;
                            this.queueOffsetMax = msg.getQueueOffset();
                        }
                    }
                    msgCount.addAndGet(validMsgCnt);
    
                    // 计算是否正在消费, 上一批次的消息是否消费完 consuming=true：还没完; consuming=false: 消费完了
                    if (!msgTreeMap.isEmpty() && !this.consuming) {
                        dispatchToConsume = true;
                        this.consuming = true;
                    }
    
                    // Broker累计消息数量
                    if (!msgs.isEmpty()) {
                        MessageExt messageExt = msgs.get(msgs.size() - 1);
                        String property = messageExt.getProperty(MessageConst.PROPERTY_MAX_OFFSET);
                        if (property != null) {
                            long accTotal = Long.parseLong(property) - messageExt.getQueueOffset();
                            if (accTotal > 0) {
                                this.msgAccCnt = accTotal;
                            }
                        }
                    }
                } finally {
                    this.lockTreeMap.writeLock().unlock();
                }
            } catch (InterruptedException e) {
                log.error("putMessage exception", e);
            }
    
            return dispatchToConsume;
        }
    
    }

“consuming” 这个变量在有序模式下，从TreeMap提取消息时，如果TreeMap为空时，会重置其为false

public class ProcessQueue { 
    
        /** * 按顺序依次弹出TreeMap的第一条消息,直到凑满batchSize * 弹出一条TreeMap就删除这条，然后重构,在弹出过程中会持有TreeMap的写锁,阻止TreeMap的写入 * 将得到的消息集合放入{@link #msgTreeMapTemp} * 当msgTreeMap里面没有消息时,重置{@link #consuming = false},也就是说现有的消息都消费完了 * * @param batchSize 条数 * @return 消息 */
        public List<MessageExt> takeMessags(final int batchSize) {
            List<MessageExt> result = new ArrayList<>(batchSize);
            final long now = System.currentTimeMillis();
            try {
                this.lockTreeMap.writeLock().lockInterruptibly();
                this.lastConsumeTimestamp = now;
                try {
                    if (!this.msgTreeMap.isEmpty()) {
                        for (int i = 0; i < batchSize; i++) {
                            Map.Entry<Long, MessageExt> entry = this.msgTreeMap.pollFirstEntry();
                            if (entry != null) {
                                result.add(entry.getValue());
                                msgTreeMapTemp.put(entry.getKey(), entry.getValue());
                            } else {
                                break;
                            }
                        }
                    }
                    //标识ProcessQueue里的msgTreeMap已经没有消息了,上一次拉取的消息已经消费完了
                    if (result.isEmpty()) {
                        consuming = false;
                    }
                } finally {
                    this.lockTreeMap.writeLock().unlock();
                }
            } catch (InterruptedException e) {
                log.error("take Messages exception", e);
            }
    
            return result;
        }
    
    }

有序模式会根据dispathToConsume 的值来决定是否提起消费请求，如果dispathToConsume=false，也就是说上一批次的消费还没有消费完，也就说有一个消费线程正在消费上批次的消息，就不起另一个线程来消费了，这样能保证消息的有序性。

public class DefaultMQPushConsumerImpl implements MQConsumerInner { 
    
        /** * 拉取消息 * * @param pullRequest 拉取消息请求 */
        public void pullMessage(final PullRequest pullRequest) {
            ......
            boolean dispathToConsume = processQueue.putMessage(pullResult.getMsgFoundList());
            // 在有序消费模式下,仅当 dispathToConsume=true 时提交消费请求,也就是上一批次的消息消费完了才提交消费请求
            // 在并发消费模式下,dispathToConsume不起作用,直接提交消费请求
            DefaultMQPushConsumerImpl.this.consumeMessageService
                .submitConsumeRequest(pullResult.getMsgFoundList(), processQueue, pullRequest.getMessageQueue(), dispathToConsume);
            ......
    
        }
    
    }
    
    public class ConsumeMessageOrderlyService implements ConsumeMessageService { 
    
        /** * 当 dispathToConsume=true 时提交消费请求,不指定拉取的消息,仅指明MessageQueue,ProcessQueue * 因此,这个消费请求会一直消费,直到{@link ProcessQueue#msgTreeMap}里没有消息 * 可能出现这种情况,消息消费的速度慢于拉取的速度,那么一个消费请求会一直持续消费 * 也就是一个线程一直维持着消费消息,不释放MessageQueue的{@link MessageQueueLock}锁 * 其他线程干瞪眼等待 * * @param msgs * @param processQueue * @param messageQueue * @param dispathToConsume */
        @Override
        public void submitConsumeRequest(final List<MessageExt> msgs,
                                         final ProcessQueue processQueue,
                                         final MessageQueue messageQueue,
                                         final boolean dispathToConsume) {
            if (dispathToConsume) {
                ConsumeRequest consumeRequest = new ConsumeRequest(processQueue, messageQueue);
                this.consumeExecutor.submit(consumeRequest);
            }
        }
    
    }

并发模式的请求对象ConsumeRequest和有序模式下的请求对象有明显不同：

public class ConsumeMessageOrderlyService implements ConsumeMessageService { 
    
        class ConsumeRequest implements Runnable {
            /** * 消息处理队列 */
            private final ProcessQueue processQueue;
            /** * 消息队列 */
            private final MessageQueue messageQueue;
        }
    
    }
    
    public class ConsumeMessageConcurrentlyService implements ConsumeMessageService { 
    
        class ConsumeRequest implements Runnable {
            /** * 消费消息列表 */
            private final List<MessageExt> msgs;
            /** * 消息处理队列 */
            private final ProcessQueue processQueue;
            /** * 消息队列 */
            private final MessageQueue messageQueue;
    
        }
    
    }

并发模式的消费请求持有需要被消费的消息，提交消费请求会根据 “consumeMessageBatchMaxSize”的大小来拆分成多个ConsumeRequest 对象，然后一起提交给消费线程池。

有序模式不持有被消费的消息，所以消费请求是直接到ProcessQueue里的TreeMap提取消息，就是上面的

public List<MessageExt> takeMessags(final int batchSize)

方法。当拉取到一批消息时，先合并入ProcessQueue的TreeMap，然后根据上批次的消息是否消费完来决定是否提交消费请求，如果没有消费完，那么继续去拉取下批次的消息(拉取前会有多个判断)。当前消费的线程会根据“consumeMessageBatchMaxSize”的大小，按顺序向TreeMap提取消息，循环消费，直到TreeMap没有剩余消息。所以可能出现一种情况，消息消费的速度 < 消息拉取速度，注意是拉取速度，不是生产速度，这种情况在有消息堆积的情况下是很正常的。此时这个线程会一直消费，每次拉取的消息合并入TreeMap中，以维持TreeMap不为空，不会有其他的消费请求被提交，也就是不会有其他的消费者线程。

**5.** 下面看有序模式的消费请求的处理：

class ConsumeRequest implements Runnable {
    
        @Override
        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;
            }
    
            // 获得 Consumer 消息队列锁，确保同一时间只有一个线程消费指定的MessageQueue
            final Object objLock = messageQueueLock.fetchLockObject(this.messageQueue);
            synchronized (objLock) {
                // (广播模式) 或者 (集群模式 && Broker消息队列锁有效)
                if (MessageModel.BROADCASTING.equals(ConsumeMessageOrderlyService.this.defaultMQPushConsumerImpl.messageModel())
                    || (this.processQueue.isLocked() && !this.processQueue.isLockExpired())) {
                    final long beginTime = System.currentTimeMillis();
                    // 循环
                    for (boolean continueConsume = true; continueConsume; ) {
                        //负载均衡后,当前MessageQueue不在属于自己,马上设置其ProcessQueue的dropped = true ,终止剩余消息的消费
                        if (this.processQueue.isDropped()) {
                            log.warn("the message queue not be able to consume, because it's dropped. {}", this.messageQueue);
                            break;
                        }
    
                        // ProcessQueue未被锁定，向Broker请求锁定相应的MessageQueue,成功了锁定ProcessQueue,然后再次消费
                        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;
                        }
                        // ProcessQueue锁过期(30S)，向Broker请求锁定相应的MessageQueue,成功了锁定ProcessQueue,然后再次消费
                        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;
                        }
    
                        // 当前轮次消费时间超过连续时长，默认：60s，提交延迟消费请求。默认情况下，每消费1分钟休息10ms。
                        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();
    
                        //按顺序提取msgTreeMap里的相应数量消息至msgTreeMapTemp临时集合中,此举会删除msgTreeMap里的那些消息，也就是这些消息目前只在临时集合msgTreeMapTemp中,之前有说明
                        List<MessageExt> msgs = this.processQueue.takeMessags(consumeBatchSize);
                        if (!msgs.isEmpty()) {
                            final ConsumeOrderlyContext context = new ConsumeOrderlyContext(this.messageQueue);
    
                            ConsumeOrderlyStatus status = null;
    
                            // Hook：before
                            ConsumeMessageContext consumeMessageContext = null;
                            if (ConsumeMessageOrderlyService.this.defaultMQPushConsumerImpl.hasHook()) {
                                consumeMessageContext = new ConsumeMessageContext();
                                consumeMessageContext
                                    .setConsumerGroup(ConsumeMessageOrderlyService.this.defaultMQPushConsumer.getConsumerGroup());
                                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 {
                                this.processQueue.getLockConsume().lock(); // 获取队列消费锁
    
                                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;
                            }
    
                            // Hook：after
                            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);
                }
            }
        }
    
    }

**final Object objLock = messageQueueLock.fetchLockObject(this.messageQueue);** ：确保同一时间只有一个线程消费指定的MessageQueue

**this.processQueue.getLockConsume().lock()** ，在执行消费方法时时锁定ProcessQueue的，所以负载均衡后，若此ProcessQueue不在属于自己，释放MessageQueu的锁前需要先获取到ProcessQueue的锁，尝试时间为1S，如果消费方法的执行时间超过1S，则可能获取其锁失败，那么MessageQueue只能等待锁时间失效了，而非主动释放。

**6.** 处理消费结果：

public class ConsumeMessageOrderlyService implements ConsumeMessageService { 
    
        public boolean processConsumeResult(final List<MessageExt> msgs,
                                            final ConsumeOrderlyStatus status,
                                            final ConsumeOrderlyContext context,
                                            final ConsumeRequest consumeRequest
        ) {
            boolean continueConsume = true;
            long commitOffset = -1L;
            if (context.isAutoCommit()) {            //默认 AutoCommit = true
                switch (status) {
                    case COMMIT:
                    case ROLLBACK:
                        log.warn("the message queue consume result is illegal, we think you want to ack these message {}", consumeRequest.getMessageQueue());
                    case SUCCESS:
                        // 提交消息已消费成功到消息处理队列
                        commitOffset = consumeRequest.getProcessQueue().commit();
                        // 统计
                        this.getConsumerStatsManager().incConsumeOKTPS(consumerGroup, consumeRequest.getMessageQueue().getTopic(), msgs.size());
                        break;
                    case SUSPEND_CURRENT_QUEUE_A_MOMENT:
                        // 统计
                        this.getConsumerStatsManager().incConsumeFailedTPS(consumerGroup, consumeRequest.getMessageQueue().getTopic(), msgs.size());
                        // 统计是否对这些消息重消费,只要重消费次数未达到最大值,都会返回true
                        if (checkReconsumeTimes(msgs)) {
                            // 设置消息重新消费,将{@link #msgTreeMapTemp}里的消息重新放回{@link #msgTreeMap},这样重新消费时就能再次消费这些消息
                            consumeRequest.getProcessQueue().makeMessageToCosumeAgain(msgs);
                            // 提交延迟消费请求,在1S后重新开启消费
                            this.submitConsumeRequestLater(consumeRequest.getProcessQueue(), consumeRequest.getMessageQueue(),
                                context.getSuspendCurrentQueueTimeMillis());
                            continueConsume = false;
                        } else {
                            commitOffset = consumeRequest.getProcessQueue().commit();
                        }
                        break;
                    default:
                        break;
                }
            } 
    
            ......
    
            // 消息处理队列未dropped，提交有效消费进度
            if (commitOffset >= 0 && !consumeRequest.getProcessQueue().isDropped()) {
                this.defaultMQPushConsumerImpl.getOffsetStore().updateOffset(consumeRequest.getMessageQueue(), commitOffset, false);
            }
    
            return continueConsume;
        }
    
    
    }

**7**. 计算消费进度，清空消费过的消息：

public class ProcessQueue { 
    
        /** * 计算消费进度 * * @return 消费进度 */
        public long commit() {
            try {
                this.lockTreeMap.writeLock().lockInterruptibly();
                try {
                    // 消费进度
                    Long offset = this.msgTreeMapTemp.lastKey();
    
                    msgCount.addAndGet(this.msgTreeMapTemp.size() * (-1));
                    //清空临时消息
                    this.msgTreeMapTemp.clear();
    
                    // 返回消费进度
                    if (offset != null) {
                        return offset + 1;
                    }
                } finally {
                    this.lockTreeMap.writeLock().unlock();
                }
            } catch (InterruptedException e) {
                log.error("commit exception", e);
            }
    
            return -1;
        }
    
    }

**8.**处理消费失败的消息：

/** * 计算是否要暂停消费 * 不暂停条件：存在消息都超过最大消费次数并且都发回broker成功 * * @param msgs 消息 * @return 是否要暂停 */
        private boolean checkReconsumeTimes(List<MessageExt> msgs) {
            boolean suspend = false;
            if (msgs != null && !msgs.isEmpty()) {
                for (MessageExt msg : msgs) {
                    if (msg.getReconsumeTimes() >= getMaxReconsumeTimes()) {
                        MessageAccessor.setReconsumeTime(msg, String.valueOf(msg.getReconsumeTimes()));
                        if (!sendMessageBack(msg)) {
                            suspend = true;
                            msg.setReconsumeTimes(msg.getReconsumeTimes() + 1);
                        }
                    } else {
                        suspend = true;
                        msg.setReconsumeTimes(msg.getReconsumeTimes() + 1);
                    }
                }
            }
            return suspend;
        }

在默认情况下， checkReconsumeTimes（msgs） == true ，也就是将消费失败的消息添加会TreeMap，在1S后重新消费这些消息

**总结：**

1.  有序模式相比较于并发模式，多个几个锁，MessageQueue的分布式锁，MessageQueue的本地锁，ProcessQueue的消费锁，这几个锁的目的都是为了**确保同一时间只有一个消费者，一个消费线程在消费指定的MessageQueue**。
2.  有序模式在负载均衡后，MessageQueue的订阅情况发生变化，消费者可能不会马上释放之前的MessageQueue的分布式锁，也就是说新的MessageQueue可能不会马上获取到锁，也就是说**第一次拉取消息相较于负载均衡有一定的延迟性**，这么做的目的是为了保证消息的有序性，不至于发生两个消费者同时消费一个MessageQueue的情况。
3.  有序模式相较于并发模式，在拉取到消息后，不一定会提交消费请求，提交与否取决于ProcessQueue里是否还有上一批次的消息未消费完。若是，则仅合并消息，不提交消费请求。提交消费请求意味着向消费线程池里提交一个ConsumeRequest，若两个线程一起消费，就不能保证消息的有序性，所以**RocketMQ在有序模式下，尽量保证同一时间只有一个线程在消费指定的MessageQueue**。
4.  同一个队列里的消息在有序消费模式下是严格有序的，当希望某一类消息严格有序时，将其发送到同一个MessageQueue下；若某一类消息只要求局部有序，比如一个消息组合由5个消息个体组成，要求这5个消息严格有序，那么通过MessageQueueSelector将其发送到同一个MessageQueue下，就能实现局部有序。**严格有序下所有消息要求发送到同一个MessageQueue，局部有序仅要求消息组合发送到同一个MessageQueue下。**局部有序可以有多个MessageQueue，而严格有序只能有一个MessageQueue。