3分钟白话RocketMQ系列—— 如何消费消息-技术圈

白话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 消费都是由客户端完成重平衡。本文还是介绍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等信息，本地计算负载均衡策略，存在消息重复的可能性。
消息消费：「消息确认机制」和「失败重试机制」 保证消息不丢失、消息队列都存在重复消费。

3分钟到了吗？应该对RocketMQ如何消费消息有全面了解了吧。
如果还想了解更多，欢迎关注下一期内容。

往期热门笔记合集推荐：

HBase原理与实战笔记合集
MySQL实战笔记合集
Canal/Otter源码与实战笔记合集
Java实战技巧笔记合集

原创：阿丸笔记（微信公众号：aone_note），欢迎分享，转载请保留出处。

没有留言功能的悲伤，扫描下方二维码「加我」聊些有的没的吧～

觉得不错，就点个再看吧👇