=
白话3分钟，快速了解RocketMQ如何消费消息。
看完如果不了解，欢迎来打我。
我们知道RocketMQ主要分为消息 生产、存储（消息堆积）、消费 三大块领域。
前面已经介绍了 生产消息、存储消息 两大块内容，那接下来，我们白话一下RocketMQ是如何消费消息的，揭秘消息消费全过程。
注意，如果白话中不小心提到相关代码配置与类名，请参考RocketMQ 4.9.4版本
关键字摘要
- 核心概念：消费者与消费组、订阅关系、消费模式
- 核心流程：消费拉取、负载均衡、消息消费
Q1: 消息消费有哪些核心概念？
消费者与消费组、订阅关系
1）消费者与消费组
消息消费以 组 的模式开展。每个消费组ConsumerGroup可以包含多个消费者Consumer，并且可以订阅多个主题Topic。如果多个消费者设置了相同的ConsumerGroup，我们认为这些消费者在同一个消费组ConsumerGroup内。
2）订阅关系
订阅关系Subscription由消费者组ConsumerGroup动态注册到服务端系统，并在后续的消息传输中按照订阅关系中的过滤规则进行 消息过滤与匹配。
原则：
- 不同消费组
ConsumerGroup
- 对于同一个
Topic
- 的订阅相互独立
- 同一个消费组
ConsumerGroup
- 对于不同
Topic
- 的订阅也相互独立
- 同一消费组
ConsumerGroup
- 内的多个消费者
Consumer
- 的订阅关系必须保持一致！否则可能会导致部分消息消费不到
3）消费模式
消费组之间有两种消费模式：「集群模式」和「广播模式」。
在「集群模式」下，同一主题下的消息只能被消费组内的某一个消费者处理，一条消息会被 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
消费端的负载均衡是指将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等信息，本地计算负载均衡策略，存在消息重复的可能性。
- 消息消费：「消息确认机制」和「失败重试机制」 保证消息不丢失、消息队列都存在重复消费。
3分钟到了吗？应该对RocketMQ如何消费消息有全面了解了吧。
如果还想了解更多，欢迎关注下一期内容。
阿丸笔记（微信公众号：aone_note）