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

【资料图】

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

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

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

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

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

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

原则：

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

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

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

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

整体流程包括：

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

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

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

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

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

嗯，一定不会啦。

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

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

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

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

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

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

负载均衡基本流程：

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

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

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

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

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

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

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

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

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

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

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

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

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

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