【深入浅出RocketMQ原理及实战】「底层原理挖掘系列」透彻剖析贯穿RocketMQ的消息顺序消费和并发消费机制体系的原理分析

简介: 【深入浅出RocketMQ原理及实战】「底层原理挖掘系列」透彻剖析贯穿RocketMQ的消息顺序消费和并发消费机制体系的原理分析

DefaultMQPushConsumerImpl拉取消息

首先,DefaultMQPushConsumerImpl 是一个实现了 RocketMQ 的消费者客户端接口的类。该类的主要作用是从 RocketMQ 的 Broker 获取消息并进行消费。

主要可以通过pullMessage方法进行获取对应的操作,如下图所示。



在消费消息时,DefaultMQPushConsumerImpl 会将获取到的消息放入一个processQueue中,processQueue包含了一个TreeMap数据结构,它按照消息的 commitLogOffset 顺序来排列。



DefaultMQPushConsumerImpl 通过定时的方式,从 Broker 上拉取消息。具体来说,它会调用DefaultMQPushConsumerImpl 自身定义的PullMessageService类,该类会定时的从消息服务器中拉取消息。


源码如下所示。



一旦消息拉取成功,PushConsumer 会将消息交给 processQueue 中的一个队列进行处理,这个队列对应同一个消息主题的同一个消息队列。

processQueue 中的每个消息都会根据消息的commitLogOffset排列位置。这个位置决定了消息被消费的顺序。也就是说,processQueue 存放的顺序决定了消息消费的顺序。

org.apache.rocketmq.client.impl.consumer.DefaultMQPushConsumerImpl#pullMessage

java

复制代码

boolean dispatchToConsume = processQueue.putMessage(pullResult.getMsgFoundList());
DefaultMQPushConsumerImpl.this.consumeMessageService.submitConsumeRequest(
    pullResult.getMsgFoundList(),
    processQueue,
    pullRequest.getMessageQueue(),
    dispatchToConsume);

consumeMessageService的并发消费和顺序消费

consumeMessageService 是一个用于消费消息的服务方法,它可以实现消息的并发消费和顺序消费。当使用 consumeMessageService 时,需要考虑业务的实际需求以及消息处理的性质,权衡使用并发消费和顺序消费。

并发消费

并发消费是指多个消费者同时消费同一批消息以提高处理速度,需要注意消息幂等性以避免重复消费。

DefaultMQPushConsumer的consumeMessageBatchMaxSize参数默认值为1,表示默认批量消费的消息数量是1个。在并发消费方式下,若一个队列中拉取到32条消息,则会创建32个ConsumeRequest对象,每个ConsumeRequest对象对应1条消息,提交到线程池中运行。

顺序消费

顺序消费则是按照消息产生的顺序逐个消费,适合处理需要顺序进行的业务逻辑,如订单处理,但实现可能带来性能瓶颈,需谨慎设计。指同一时刻,一个 queue 只有一个线程在消费。只让一个线程消费,由加锁来实现,而顺序则由 TreeMap 来实现

一个队列中拉取到32条消息,则只会创建一个ConsumeRequest对象,该对象会被提交到线程池中,在ConsumeRequest.run方法中会按照消息的offset顺序一条一条地消费,直到TreeMap为空

concurrently 创建 ConsumeRequest

java

复制代码

public void submitConsumeRequest(
    final List<MessageExt> msgs,
    final ProcessQueue processQueue,
    final MessageQueue messageQueue,
    final boolean dispatchToConsume) {
    final int consumeBatchSize = this.defaultMQPushConsumer.getConsumeMessageBatchMaxSize();
    if (msgs.size() <= consumeBatchSize) {
        ConsumeRequest consumeRequest = new ConsumeRequest(msgs, processQueue, messageQueue);
        try {
            this.consumeExecutor.submit(consumeRequest);
        } catch (RejectedExecutionException e) {
            this.submitConsumeRequestLater(consumeRequest);
        }
    } else {
        for (int total = 0; total < msgs.size(); ) {
            List<MessageExt> msgThis = new ArrayList<MessageExt>(consumeBatchSize);
            for (int i = 0; i < consumeBatchSize; i++, total++) {
                if (total < msgs.size()) {
                    msgThis.add(msgs.get(total));
                } else {
                    break;
                }
            }
            ConsumeRequest consumeRequest = new ConsumeRequest(msgThis, processQueue, messageQueue);
            try {
                this.consumeExecutor.submit(consumeRequest);
            } catch (RejectedExecutionException e) {
                for (; total < msgs.size(); total++) {
                    msgThis.add(msgs.get(total));
                }
                this.submitConsumeRequestLater(consumeRequest);
            }
        }
    }
}

消费者在消费消息时,根据批量消费的大小来决定是将任务提交到线程池中一次性消费,还是将任务分成多次提交到线程池中进行消费。

首先判断msgs中消息的数量是否小于等于一个批量消费数量consumeBatchSize,如果小于等于,那么将所有消息封装成一个ConsumeRequest对象并提交到consumeExecutor线程池中,其中dispatchToConsume表示是否立即分发给消费者消费。

如果消息数量大于批量消费数量,那么将消息分段提交到线程池中进行消费。首先通过两层循环,将msgs中的消息按照consumeBatchSize分成若干个小的MessageExt列表,每个小的MessageExt列表封装成一个ConsumeRequest对象并提交到consumeExecutor线程池中。

如果线程池提交任务出现拒绝执行异常,说明该线程池已经满了,这时候需要将当前小的MessageExt列表继续循环并依次每次取出一个消息封装成ConsumeRequest对象进行提交,直到所有的小的MessageExt列表被完整地提交到线程池中。若还有未提交的列表,则将该ConsumeRequest对象提交到一个新的线程池中进行定时的重复提交。

concurrently ConsumeRequest#run 消费主体逻辑

消息消费者消费消息的地方,listener.consumeMessage方法会被消费者调用,将消息列表和消息处理上下文传入。

java

复制代码

status = listener.consumeMessage(Collections.unmodifiableList(msgs), context);
  • msgs是需要消费的消息列表,这里使用了Collections.unmodifiableList方法来创建一个不可修改的消息列表,这是为了保证消息的安全性,防止消息在消费过程中被意外或恶意修改。
  • context是消息处理的上下文,可能包含消费者的订阅信息、消费进度等信息,可根据业务需要进行扩展和使用。
  • consumeMessage方法返回消费结果,通常是一个枚举类型,表示消费结果的状态,如消费成功、消费失败等。消费结果会影响消息处理的下一步流程。
消费结束之后清除数据

主要用于移除已经消费完成的消息。直接从 msgTreeMap 中删除消息,并返回 msgTreeMap 中第一条消息的 queue offset 值。

org.apache.rocketmq.client.impl.consumer.ProcessQueue#removeMessage

java

复制代码

public long removeMessage(final List<MessageExt> msgs) {
    long result = -1;
    final long now = System.currentTimeMillis();
    try {
        this.lockTreeMap.writeLock().lockInterruptibly();
        this.lastConsumeTimestamp = now;
        try {
            if (!msgTreeMap.isEmpty()) {
                result = this.queueOffsetMax + 1;
                int removedCnt = 0;
                for (MessageExt msg : msgs) {
                    MessageExt prev = msgTreeMap.remove(msg.getQueueOffset());
                    if (prev != null) {
                        removedCnt--;
                        msgSize.addAndGet(0 - msg.getBody().length);
                    }
                }
                msgCount.addAndGet(removedCnt);
                if (!msgTreeMap.isEmpty()) {
                    result = msgTreeMap.firstKey();
                }
            }
        } finally {
            this.lockTreeMap.writeLock().unlock();
        }
    } catch (Throwable t) {
        log.error("removeMessage exception", t);
    }
    return result;
}

具体来说,它接收一个 MessageExt 类型的消息列表msgs,通过遍历msgs,查找msgTreeMap中相应的消息,将找到的消息删除并计数,更新msgCount和msgSize这两个计数器。代码中也使用了重入锁lockTreeMap来保证线程安全。函数将返回result,表示下一步应该消费的消息的offset,如果没有可消费的消息,则返回-1。

orderly 创建 ConsumeRequest

在消息消费过程中,判断是否需要立即将消息分发给消费者进行消费。

java

复制代码

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);
    }
}

首先判断参数dispathToConsume为true,如果为true,表示需要立即分发给消费者消费;否则就不需要进行分发,因为可能等待其他条件触发再进行消费。

如果需要立即分发,那么将该消息的消息队列和消息处理队列封装成ConsumeRequest对象,并将该对象提交到consumeExecutor线程池中进行执行。每个消费者线程从consumeExecutor线程池中取出ConsumeRequest对象并进行消费。

orderly ConsumeRequest#run 消费主体逻辑

先简单介绍一下 RocketMQ 消息消费的流程:消费者将消息从 Broker 中拉取到本地的 ProcessQueue 中,然后在 ProcessQueue 中进行消息消费。

java

复制代码

// 获取锁
final Object objLock = messageQueueLock.fetchLockObject(this.messageQueue);
synchronized (objLock) {
    for (boolean continueConsume = true; continueConsume; ) {
        // 从 TreeMap 中获得消息
        List<MessageExt> msgs = this.processQueue.takeMessags(consumeBatchSize);
        if (!msgs.isEmpty()) {
            status = messageListener.consumeMessage(Collections.unmodifiableList(msgs), context);
        } else {
            continueConsume = false;
        }
    }
    ...
}
public class MessageQueueLock {
    private ConcurrentMap<MessageQueue, Object> mqLockTable = new ConcurrentHashMap<MessageQueue, Object>();
    public Object fetchLockObject(final MessageQueue mq) {
        Object objLock = this.mqLockTable.get(mq);
        if (null == objLock) {
            objLock = new Object();
            Object prevLock = this.mqLockTable.putIfAbsent(mq, objLock);
            if (prevLock != null) {
                objLock = prevLock;
            }
        }
        return objLock;
    }
}

首先实例化了 MessageQueueLock,用于保证多线程环境下的线程同步和互斥。在代码的第一行中,获取到了当前 MessageQueue 的锁对象 objLock。这个锁对象是在 mqLockTable 中获取的,mqLockTable 存储了每个 MessageQueue 的锁对象,用于对不同的 MessageQueue 进行互斥控制。

在代码的后面,使用 synchronized 对 objLock 进行加锁,并进入到了循环中。在循环中,调用 processQueue.takeMessags() 方法从 ProcessQueue 中获取消息,返回的是一个消息列表。如果消息列表不为空,则调用 messageListener.consumeMessage() 方法来进行消息消费。

如果消息列表为空,说明当前的 ProcessQueue 中没有更多的消息,结束当前的循环,并退出 synchronized 块,释放了 objLock 的锁,等待下一次的消费请求。

整个逻辑是通过锁机制来实现对 ProcessQueue 进行互斥控制的,保证了多个消费者之间的消费的安全性。同时,使用了循环来进行多次消费。

顺序处理机制

take消息时,将消息从 msgTreeMap 取出,并放入 consumingMsgOrderlyTreeMap。消费完成后,清空 consumingMsgOrderlyTreeMap。将 offset 设为 this.consumingMsgOrderlyTreeMap.lastKey() + 1,表示已经消费的消息的下一条消息的 offset。

// org.apache.rocketmq.client.impl.consumer.ProcessQueue#commit

java

复制代码

public long commit() {
    try {
        this.lockTreeMap.writeLock().lockInterruptibly();
        try {
            Long offset = this.consumingMsgOrderlyTreeMap.lastKey();
            msgCount.addAndGet(0 - this.consumingMsgOrderlyTreeMap.size());
            for (MessageExt msg : this.consumingMsgOrderlyTreeMap.values()) {
                msgSize.addAndGet(0 - msg.getBody().length);
            }
            this.consumingMsgOrderlyTreeMap.clear();
            if (offset != null) {
                return offset + 1;
            }
        } finally {
            this.lockTreeMap.writeLock().unlock();
        }
    } catch (InterruptedException e) {
        log.error("commit exception", e);
    }
    return -1;
}
关于 offset 提交

offset 是消费者从 broker 拉取的下一条消息的偏移量

消息消费的失败

  • 顺序消费:如果处理某条消息失败且重试次数小于阈值,从 consumingMsgOrderlyTreeMap 中取出这条消息并重新放入 msgTreeMap;如果重试次数超过阈值,则将消息发送回 broker 并根据重试次数决定发送消息到 SCHDULE_TOPIC_XXXX 或死信队列
  • 并发消费:如果处理消息时失败,则将消息发送回 broker。如果发送失败,将会继续消费消息,直到成功消费并提交给 broker。

发送 ConsumeRequest 的时机有两个,一是在拉取到消息后,二是在出现异常后延迟提交。

相关实践学习
快速体验阿里云云消息队列RocketMQ版
本实验将带您快速体验使用云消息队列RocketMQ版Serverless系列实例进行获取接入点、创建Topic、创建订阅组、收发消息、查看消息轨迹和仪表盘。
消息队列 MNS 入门课程
1、消息队列MNS简介 本节课介绍消息队列的MNS的基础概念 2、消息队列MNS特性 本节课介绍消息队列的MNS的主要特性 3、MNS的最佳实践及场景应用 本节课介绍消息队列的MNS的最佳实践及场景应用案例 4、手把手系列:消息队列MNS实操讲 本节课介绍消息队列的MNS的实际操作演示 5、动手实验:基于MNS,0基础轻松构建 Web Client 本节课带您一起基于MNS,0基础轻松构建 Web Client
相关文章
|
1月前
|
消息中间件 Ubuntu Java
SpringBoot整合MQTT实战:基于EMQX实现双向设备通信
本教程指导在Ubuntu上部署EMQX 5.9.0并集成Spring Boot实现MQTT双向通信,涵盖服务器搭建、客户端配置及生产实践,助您快速构建企业级物联网消息系统。
373 1
|
4月前
|
消息中间件 存储 Kafka
一文带你从入门到实战全面掌握RocketMQ核心概念、架构部署、实践应用和高级特性
本文详细介绍了分布式消息中间件RocketMQ的核心概念、部署方式及使用方法。RocketMQ由阿里研发并开源,具有高性能、高可靠性和分布式特性,广泛应用于金融、互联网等领域。文章从环境搭建到消息类型的实战(普通消息、延迟消息、顺序消息和事务消息)进行了全面解析,并对比了三种消费者类型(PushConsumer、SimpleConsumer和PullConsumer)的特点与适用场景。最后总结了使用RocketMQ时的关键注意事项,如Topic和Tag的设计、监控告警的重要性以及性能与可靠性的平衡。通过学习本文,读者可掌握RocketMQ的使用精髓并灵活应用于实际项目中。
2739 9
 一文带你从入门到实战全面掌握RocketMQ核心概念、架构部署、实践应用和高级特性
|
5月前
|
消息中间件 架构师 Java
美团面试:对比分析 RocketMQ、Kafka、RabbitMQ 三大MQ常见问题?
美团面试:对比分析 RocketMQ、Kafka、RabbitMQ 三大MQ常见问题?
美团面试:对比分析 RocketMQ、Kafka、RabbitMQ 三大MQ常见问题?
|
6月前
|
消息中间件 NoSQL Java
RocketMQ实战—10.营销系统代码优化
本文主要介绍了如何对营销系统的四大促销场景的代码进行优化,包括:全量用户推送促销活动、全量用户发放优惠券、特定用户推送领取优惠券消息、热门商品定时推送。
|
3月前
|
消息中间件 数据管理 Serverless
阿里云消息队列 Apache RocketMQ 创新论文入选顶会 ACM FSE 2025
阿里云消息团队基于 Apache RocketMQ 构建 Serverless 消息系统,适配多种主流消息协议(如 RabbitMQ、MQTT 和 Kafka),成功解决了传统中间件在可伸缩性、成本及元数据管理等方面的难题,并据此实现 ApsaraMQ 全系列产品 Serverless 化,助力企业提效降本。
|
1月前
|
消息中间件 Java Kafka
消息队列比较:Spring 微服务中的 Kafka 与 RabbitMQ
本文深入解析了 Kafka 和 RabbitMQ 两大主流消息队列在 Spring 微服务中的应用与对比。内容涵盖消息队列的基本原理、Kafka 与 RabbitMQ 的核心概念、各自优势及典型用例,并结合 Spring 生态的集成方式,帮助开发者根据实际需求选择合适的消息中间件,提升系统解耦、可扩展性与可靠性。
126 1
消息队列比较:Spring 微服务中的 Kafka 与 RabbitMQ
|
12月前
|
消息中间件 JSON Java
开发者如何使用轻量消息队列MNS
【10月更文挑战第19天】开发者如何使用轻量消息队列MNS
787 104
|
12月前
|
消息中间件 安全 Java
云消息队列RabbitMQ实践解决方案评测
一文带你详细了解云消息队列RabbitMQ实践的解决方案优与劣
307 111
|
11月前
|
消息中间件 存储 Kafka
MQ 消息队列核心原理,12 条最全面总结!
本文总结了消息队列的12个核心原理,涵盖消息顺序性、ACK机制、持久化及高可用性等内容。关注【mikechen的互联网架构】,10年+BAT架构经验倾囊相授。
|
12月前
|
消息中间件
解决方案 | 云消息队列RabbitMQ实践获奖名单公布!
云消息队列RabbitMQ实践获奖名单公布!
177 1