我们知道RocketMQ主要分为消息生产、存储（消息堆积）、消费三大块领域。

【资料图】

前面已经介绍了 生产消息、存储消息 两大块内容，那接下来，我们白话一下RocketMQ是如何消费消息的，揭秘消息消费全过程。

注意，如果白话中不小心提到相关代码配置与类名，请参考RocketMQ 4.9.4版本

关键字摘要核心概念：消费者与消费组、订阅关系、消费模式核心流程：消费拉取、负载均衡、消息消费Q1: 消息消费有哪些核心概念？

消费者与消费组、订阅关系

消费者与消费组

消息消费以 组 的模式开展。每个消费组ConsumerGroup可以包含多个消费者Consumer，并且可以订阅多个主题Topic。

如果多个消费者设置了相同的ConsumerGroup，我们认为这些消费者在同一个消费组ConsumerGroup内。

订阅关系

订阅关系Subscription由消费者组ConsumerGroup动态注册到服务端系统，并在后续的消息传输中按照订阅关系中的过滤规则进行消息过滤与匹配。

原则：

不同消费组ConsumerGroup对于同一个Topic的订阅相互独立。同一个消费组ConsumerGroup对于不同Topic的订阅也相互独立。同一消费组ConsumerGroup内的多个消费者Consumer的订阅关系必须保持一致！否则可能会导致部分消息消费不到。消费模式

消费组之间有两种消费模式：「集群模式」和「广播模式」。

在「集群模式」下，同一主题下的消息只能被消费组内的某一个消费者处理，一条消息会被 1 个消费组内的 N 个消费者消费 1 次。

在「广播模式」下，同一主题下的消息将会被消费组内的所有消费者处理一次，一条消息会被 1 个消费组内的 N 个消费者消费 N 次。

如果消息消费是「集群模式」，那么消息进度保存在Broker上; 如果是「广播模式」，那么消息消费进度存储在Consumer端本地。

Q2：消费者怎么拉取消息？

整体流程包括：

消费者启动。主要包括订阅Topic、初始化消息进度。消费者发送拉取请求。主要查询路由表找到目标Broker发送请求。Broker查找并返回消息。根据订阅关系Subscription和 消息进度 进行消息过滤和匹配，然后返回消息。消费者接收并处理消息。

消息服务器与消费者之间有两种消息传送方式：「推模式」和「拉模式」。

「拉模式」是消费者主动向消息服务器请求拉取消息。「推模式」是消息到达消息服务器后，由服务器主动推送给消息消费者。

在 RocketMQ 中，Consumer端的两种消费模式（Push/Pull）底层其实都是基于「拉模式」来获取消息的。

具体实现方式是，消息拉取线程从服务器拉取一批消息后，将其提交给消息消费线程池，并立即继续向服务器尝试拉取消息，以保持消息的连续性。

那如果拉取消息时，Broker端暂时没有新消息可以返回怎么办？会一直无脑发送拉取请求吗？

嗯，一定不会啦。

RocketMQ默认会开启「长轮询机制」，这个机制能够平衡轮询压力与新消息的实时性：

消费者发送拉取请求到Broker，如果没有新消息，Broker会暂时 挂起 请求不返回。Broker每隔5s检查一次挂起的请求，是否有满足条件的新消息，如果有就返回，如果没有就继续挂起，直到超时返回。如果在挂起的过程中，有满足条件的新消息写入commitLog，也会立即返回新消息。Q3：消费者怎么知道去哪里拉取消息？

这就需要聊一聊消息消费的「负载均衡机制」了。

注意，RocketMQ 5.x版本，对「推模式」底层增加了一种「Pop模式」的实现。Pop和Pull区别在于，Pop消费的重平衡是在 Broker 端做的，而之前的Pull消费都是由客户端完成重平衡。本文还是介绍4.x版本。

消费端的负载均衡是指将Broker端中多个队列queue按照某种算法分配给同一个消费组中的不同消费者，负载均衡是客户端开始消费的起点。

注意，从RocketMQ服务端5.0版本开始额外支持了「消息粒度」的负载均衡策略，4.x/3.x版本仅支持「队列粒度」的负载均衡策略。本文只介绍4.x的「队列粒度」的。

RocketMQ「队列粒度」的负载均衡的核心设计理念是：

消费队列在同一时间只允许被同一消费组内的一个消费者消费一个消费者能同时消费多个消息队列

负载均衡基本流程：

Consumer启动后，它就会通过定时任务向所有Broker实例发送心跳包（包含消费分组名称、订阅关系集合、消息通信模式和客户端id等信息），Broker会缓存这些信息。Consumer每隔10ms从Nameserver获取Topic与队列queue的路由信息，缓存本地每隔20s，Consumer端会请求Broekr获取该消费组下消费者Id列表，然后根据Topic下的队列queue、消费组下消费者Id进行排序，计算出待拉取的队列queue根据新算出的本地应该消费队列queue，重新计算本地队列消费任务。

特别注意，无论是消息粒度负载均衡策略还是队列粒度负载均衡策略，在消费者上线或下线、服务端扩缩容等场景下，都会触发短暂的重新负载均衡动作，可能会存在短暂的负载不一致情况，出现少量消息重复的现象。

因此，需要在下游消费逻辑中做好消息「幂等去重」处理。

Q4: 消费者拉到消息了，怎么消费呢？

消息消费，主要关注两个事情：

会不会消息丢失？会不会消费重复？

怎么保证消息消费不丢失？

其实思路是比较直接的，就是「消息确认机制」和「失败重试机制」。

消费者从RocketMQ拉取消息后，需要返回"CONSUME_SUCCESS"来表示业务方已经正常完成消费。只有返回"CONSUME_SUCCESS"才算作消费完成。这就是消费时的「消息确认机制」。

如果返回"CONSUME_LATER"，则会按照不同的消息延迟级别进行再次消费，延迟级别从秒到小时不等，最长延迟时间为2个小时后再次尝试消费。这就是消费时的「失败重试机制」。

重试消息会被存入名为"%RETRY%+消费组名称"的Topic中，原始主题Topic会存入属性中。然后会基于定时任务机制，在到期时将任务再次拉取出来。

注意，从重试Topic的名称我们可以了解到，RocketMQ消息重试是以消费组为单位，而不是Topic。

另外，RocketMQ跟kafka不同的是，天然支持了「死信队列机制」。

如果在尝试消费的过程中达到了最大重试次数（通常为16次），仍然无法成功消费，则消息将被发送到死信队列，以确保消息存储的可靠性。后续业务可以根据死信队列，来做相关补偿措施。

怎么保证消息消费不重复？

其实思路也很直接，就是不保证不重复。

所有消息队列的设计，都是不保证消息消费不重复的。所以使用消息队列时，要特别注意，如果有唯一性要求，必须做好消费端的「幂等设计」。

总结消息拉取：「推模式」与「拉模式」本质都是「拉模式」、「长轮询机制」平衡 轮询压力 与 新消息的实时性。消息消费负载均衡：定时获取Topic下的队列queue、消费组下消费者Id等信息，本地计算负载均衡策略，存在消息重复的可能性。消息消费：「消息确认机制」和「失败重试机制」保证消息不丢失、消息队列都存在重复消费。

关键词：