【rocketmq客户端】订阅关系一致

Posted 2023-04-23

参考技术A

参考：

同一个消费组中，不一致的订阅行为有：

不一致的订阅行为，会导致部分消息丢失。

rocket mq的tag级别的消息过滤功能，其实现原理包括两方面：

存储模型
rocket mq为每一个topic建立索引，方便消费者按照topic进行消费，其具体实现为ConsumeQueue(消息队列)。consumeQueue的设计极具技巧性，队列中每个条目长度固定，组成为：

这种设计使得消费队列具备类似数组般随机访问的特性，提高consumeQueue的读取性能。

tag过滤
tag消息过滤时：

消息丢失
为什么同一消费组，多个消费者订阅同一topic不同的tag，会发生消息丢失？

使用集群模式消费时，同一个消费组内的多个消费者共同完成topic的多个consumeQueue消费，一个消费组只会分配到其中某几个consumeQueue，且同一时间，一个consumeQueue只会分配给一个消费组。如图：

问题的核心是：同一个tag的消息会分布在不同的consumeQueue中，而队列的消息只会被一个消费者消费，那么这个consumerQueue上，不同于消费者订阅tag的消息，将不会被消费，造成堆积。

如下，消费组 consumer-group-name 下的两个消费者 consumer 1 consumer2 分别订阅了两个不同的topic topic A 与 topic B ：

由于consumerQueue的分配是以消费组为单位的，会均匀分配给消费组下的消费者，而不会在意消费者本身订阅的是哪个主题。

topic A的consumerQueue q4 分配给了consumer2，但是consumer2订阅的是topic B，因此部分topic A的消息得不到消费。

得出结论：

RocketMQ

消费流程

消费者组： 一个逻辑概念，在使用消费者时需要指定一个组名。一个消费者组可以订阅多个Topic。
消费者实例： 一个消费者组程序部署了多个进程，每个进程都可以称为一个消费者实例。
订阅关系： 一个消费者组订阅一个 Topic 的某一个 Tag，这种记录被称为订阅关系。RocketMQ规定消费订阅关系（消费者组名-Topic-Tag）必须一致——在此，笔者想提醒读者，一定要重视这个问题，一个消费者组中的实例订阅的Topic和Tag必须完全一致，否则就是订阅关系不一致。订阅关系不一致会导致消费消息紊乱。

消费模式

RocketMQ目前支持集群消费模式和广播消费模式，其中集群消费模式使用最为广泛

集群消费模式

在同一个消费者组中的消费者实例，是负载均衡（策略可以配置）地消费Topic中的消息，假如有一个生产者（Producer）发送了 120 条消息，其所属的 Topic 有 3 个消费者（Consumer）组，每个消费者组设置为集群消费，分别有2个消费者实例，如图所示。

Consumer Group A 的两个实例 Consumer Instance A1 和 Consumer Instance A2 分别负载均衡地消费60条消息。由此我们可以得出使用负载均衡策略时，每个消费者实例消费消息数=生产消息数/消费者实例数，在本例中是60=120/2

适用场景： 目前大部分场景都适合集群消费模式，RocketMQ 的消费模式默认是集群消费。比如异步通信、削峰等对消息没有顺序要求的场景都适合集群消费。因为集群模式的消费进度是保存在Broker端的，所以即使应用崩溃，消费进度也不会出错。

广播消费模式

广播消费，顾名思义全部的消息都是广播分发，即消费者组中的全部消费者实例将消费整个 Topic 的全部消息。比如，有一个生产者生产了 120 条消息，其所属的 Topic 有 3个消费者组，每个消费者组设置为广播消费，分别有两个消费者实例，如图所示。

Consumer Group A 的两个实例 Consumer Instance A1 和 Consumer Instance A2 分别消费120条消息。整个消费者组收到消息120×2=240条。由此我们可以得出广播消费时，每个消费者实例的消费消息数=生产者生产的消息数，整个消费者组中所有实例消费消息数=每个消费者实例消费消息数×消
费者实例数，本例中是240=120×2

适用场景： 广播消费比较适合各个消费者实例都需要通知的场景，比如刷新应用服务器中的缓存，如图所示。

生产者发一个刷新缓存的广播消息，消费者组如果设置为广播消费，那么每个应用服务中的消费者都可以消费这个消息，也都能刷新缓存。

广播消费的消费进度保存在客户端机器的文件中。如果文件弄丢了，那么消费进度就丢失了，可能会导致部分消息没有消费

可靠消费

RocketMQ是一种十分可靠的消息队列中间件，消费侧通过重试-死信机制、Rebalance机制等多种机制保证消费的可靠性。

重试-死信机制

我们假设有一个场景，在消费消息时由于网络不稳定导致一条消息消费失败。此时是让生产者重新手动发消息呢，还是自己做数据补偿？

横向看，RocketMQ的消费过程分为 3个阶段：正常消费、重试消费和死信。在引进了正常Topic、重试队列、死信队列后，消费过程的可靠性提高了。RocketMQ的消费流程如图所示

正常Topic： 正常消费者订阅的Topic名字。
重试Topic： 如果由于各种意外导致消息消费失败，那么该消息会自动被保存到重试Topic中，格式为“%RETRY%消费者组”，在订阅的时候会自动订阅这个重试Topic。

进入重试队列的消息有16次重试机会，每次都会按照一定的时间间隔进行。RocketMQ 认为消费不是一锤子买卖，可能由于各种偶然因素导致正常消费失败，只要正常消费或者重试消费中有一次消费成功，就算消费成功

死信Topic： 死信Topic名字格式为“%DLQ%消费者组名”。如果正常消费1次失败，重试16次失败，那么消息会被保存到死信Topic中，进入死信Topic的消息不能被再次消费。RocketMQ认为，如果17次机会都失败了，说明生产者发送消息的格式发生了变化，或者消费服务出现了问题，需要人工介入处理。

Rebalance机制

Rebalance（重平衡）机制，用于在发生Broker掉线、Topic扩容和缩容、消费者扩容和缩容等变化时，自动感知并调整自身消费，以尽量减少甚至避免消息没有被消费。后面会详细讲述Rebalance的过程。

死信队列

当一条消息初次消费失败，消息队列RocketMQ版会自动进行消息重试，达到最大重试次数后，若消费依然失败，则表明消费者在正常情况下无法正确地消费该消息。此时，消息队列RocketMQ版不会立刻将消息丢弃，而是将其发送到该消费者对应的特殊队列中。

在消息队列RocketMQ版中，这种正常情况下无法被消费的消息称为死信消息（Dead-Letter Message），存储死信消息的特殊队列称为死信队列（Dead-Letter Queue）

死信消息具有以下特性：

• 不会再被消费者正常消费。
• 有效期与正常消息相同，默认为3天，3天后会被自动删除。因此，请在死信消息产生后的3天内及时处理。

死信队列具有以下特性：

• 一个死信队列对应一个Group ID， 而不是对应单个消费者实例。
• 如果一个Group ID未产生死信消息，消息队列RocketMQ版不会为其创建相应的死信队列。
• 一个死信队列包含了对应Group ID产生的所有死信消息，不论该消息属于哪个Topic。