一、方案背景
RocketMQ（以下简称MQ）作为消息中间件在事务管理，异步解耦，削峰填谷，数据同步等应用场景中有着广泛使用。当业务系统进行灰度发布时，Dubbo与HTTP的调用可以基于业界通用的灰度方式在我们的微服务治理与网关平台来实现，但MQ已有的灰度方案都不能完全解决消息的隔离与切换衔接问题，为此，我们在鲁班MQ平台（包含根因分析、资源管理、订阅关系校验、延时优化等等的扩展）增加了MQ灰度功能的扩展实现。
二、RocketMQ技术特点
为什么MQ的灰度方案迟迟没有实现呢？我们先来回顾一下RocketMQ的几个核心技术点。
2.1 存储模型的简述
![图片]()
（图2.1 MQ的存储模型）
CommitLog：消息体实际存储的地方，当我们发送的任一业务消息的时候，它最终会存储在commitLog上。MQ在Broker进行集群部署（这里为也简洁，不涉及主从部分）时，同一业务消息只会落到集群的某一个Broker节点上。而这个Broker上的commitLog就会存储所有Topic路由到它的消息，当消息数据量到达1个G后会重新生成一个新的commitLog。
Topic：消息主题，表示一类消息的逻辑集合。每条消息只属于一个Topic，Topic中包含多条消息，是MQ进行消息发送订阅的基本单位。属于一级消息类型，偏重于业务逻辑设计。
Tag：消息标签，二级消息类型，每一个具体的消息都可以选择性地附带一个Tag，用于区分同一个Topic中的消息类型，例如订单Topic， 可以使用Tag=tel来区分手机订单，使用Tag=iot来表示智能设备。在生产者发送消息时，可以给这个消息指定一个具体的Tag， 在消费方可以从Broker中订阅获取感兴趣的Tag，而不是全部消息（注：严谨的拉取过程，并不全是在Broker端过滤，也有可能部分在消费方过滤，在这里不展开描述）。
Queue：实际上Topic更像是一个逻辑概念供我们使用，在源码层级看，Topic以Queue的形式分布在多个Broker上，一个topic往往包含多条Queue（注：全局顺序消息的Topic只有一条Queue，所以才能保证全局的顺序性），Queue与commitLog存在映射关系。可以理解为消息的索引，且只有通过指定Topic的具体某个Queue，才能找到消息。（注：熟悉kafka的同学可以类比partition）。
消费组及其ID：表示一类Producer或Consumer，这类Producer或Consumer通常生产或消费同应用域的消息，且消息生产与消费的逻辑一致。每个消费组可以定义全局维一的GroupID来标识，由它来代表消费组。不同的消费组在消费时互相隔离，不会影响彼此的消费位点计算。
2.2 消息发送与消费
![图片]()
（图2.2 消息发送与拉取模型）
2.2.1 客户端标识
在生产者或消费者集群中，每一个MQ客户端的运行实例，在MQ的客户端会保证产生唯一的ClientID。注：同一应用实例中，既充当生产者，也充当消费者时，其ClientID实际上是同一个取值。
2.2.2 消息发送
当向某个Topic发送消息的时候，会首先获得这个Topic的元数据，元数据会包括它有多少个Queue，每个Queue属于哪个Broker。默认的情况下，发送方会选择一条Queue发送当前消息，算法类型是轮询，也就是下一条消息会选择另一条Queue进行发送。另外MQ也提供了指定某条Queue或者自定义选择Queue的方法进行消息的发送，自定义选择Queue需实现MessageQueueSelector接口。
2.2.3 消息消费
进行消息的消费时，同一消费组(GroupID)的消费者会订阅Topic，消费者首先获得Topic的元数据，也就是会获得这个Topic的所有Queue信息。然后这些Queue按规则分配给各个具体的客户端（ClientID)，各个客户端根据分配到的Queue计算对应的需要拉取消息的offset并生成PullRequest，拉取消息并消费完成后，客户端会生成ACK并更新消费进度。
这里的消费进度是该批消息未消费成功的最小offset，如图2.3所示，一批消息中如果1、5未消费，其余的消息已消费，此时更新的offset仍是1，消费者如果宕机重启，会从1号开始消费消息，此时2、3、4号消息将会重复消费。
![图片]()
（图2.3 消费进度更新示意图）
因此RocketMQ只保证了消息不会丢失，无法保证消息不会重复消费，消息的幂等性需要业务自己实现。
另外，消费方可以指定消费某些Tag的消息，在pullRequest进行拉取时，会在Broker里会按照存储模型的Queue索引信息按hash值进行快速过滤，返回到消费方，消费方也会对返回的消息进行精准的过滤。
2.3 订阅关系一致性
在消费端，同一消费组内（GroupID相同，本节所描述的前提都是同一消费组内）的各个应用实例MQ客户端需要保持订阅关系的一致性，所谓订阅关系的一致性就是同一消费组内的所有客户端所订阅的Topic和Tag必须完全一致，如果组内订阅关系不一致，消息消费的逻辑就会混乱，甚至导致消息丢失。
2.3.1 订阅关系的维护
每一个应用实例MQ客户端都有独立的ClientID，我们简单地讲解一下订阅关系的维护：
-
各个MQ消费客户端会带着它的ClientID向所有Broker发送心跳包，心跳包中含有具体的本客户端订阅的Topic & Tag等信息，registerConsumer方法会按照消费组将客户端分组存储，同一消费组内的ClientID信息在同一个ConsumerGroupInfo中；
-
Broker在接收到任何消费客户端发过来的心跳包后，会在ConsumerManager类中的ConcurrentMapconsumerTable按照消费组名称GroupID作为key存储不同的消费组信息，同一消费组的订阅信息会在每一次收到心跳包后，都会按当前的订阅Topic & Tag信息进行更新维护，也就是相当于只保存最新的心跳包订阅信息（心跳包中的subVersion会标记心跳包版本，当重平衡结果发生改变后，subVersion会更新，Broker只会保存最新版本的心跳包中的订阅信息），不管这个心跳包来源于这个消费组的哪个ClientID。（由于Tag是依赖于Topic的属性，Topic和Tag订阅关系不一致时Broker对应的处理结果也略有不同，具体可见updateSubscription方法）。
2.3.2 订阅不一致影响
在这里使用例子的方式来阐述订阅关系不一致带来的部分问题。假设同一组内的消费者clientA订阅了TOPIC_A，clientB订阅了TOPIC_B；TOPIC_A、TOPIC_B均为4条Queue，在进行消费分配时，最终Queue的分配结果如下：
![图片]()
（表2.1 消息费者的Queue分配结果表)
因为clientB没有订阅TOPIC_A，clientA也没有订阅TOPIC_B，所以TOPIA_A中的Queue-2、Queue-3，TOPIC_B中的Queue-0、Queue-1中的消息则无法正常消费。而且在实际过程中，clientA也没法正常消费TOPIC_B的消息，这时会在客户端看到很多异常，有些Queue的消息得不到消费产生堆积。
此外，在订阅关系中，不同client所订阅的Tag不一样时，也会发生拉取的消息与所需要订阅的消息不匹配的问题。
三、业界MQ灰度方案
![图片]()
（图3.1 灰度调用示意图）
通常，业务灰度只严格地保证RPC服务之间的调用，部分消息灰度流量的流失或错误是可以容忍的，如图3-1所示，V_BFF产生的灰度消息会被V_TRADE的正常版本与灰度版本收到并随机消费，导致部分灰度流量没有进入期望的环境，但整体RPC服务的调用还是隔离了灰度与非灰度环境。当业务对消息消费的逻辑进行了更改，或者不想让灰度的消息影响线上的数据时，MQ的灰度就必须要实现。
由于订阅关系的限制，当前的业界实现的MQ灰度方案都是正常版本与灰度版本使用不同的GroupID来实现。以下的方案均使用了不同的GroupID。
3.1 影子Topic的方案
新建一系列新的Topic来处理隔离灰度的消息。例如对于TOPIC_ORDER会创建TOPIC_ORDER_GRAY来给灰度环境使用。
发送方在发送时，对灰度的消息写入到影子Topic中。消费时，灰度环境只使用灰度的分组订阅灰度的Topic。
3.2 Tag的方案
发送方在发送时，对灰度环境产生的消息的Tag加注灰度标识。消费方，每次灰度版本发布时只订阅灰度Tag的消息，正常的版本订阅非灰度的Tag。
3.3 UserProperty的方案
发送方在发送时，对灰度环境产生的消息的UserProperty加注灰度标识。消费方的客户端需要进行改写，根据UserProperty来过滤，正常的版本会跳过这类消息，灰度的版本会处理灰度消息。
3.4 当前的方案缺陷
以上三种方案各自的优势在这里不做比较，但都存在以下共同的缺陷（也有其它的缺陷或开发诉求，但不致命），无法真正实现灰度状态切换回正常状态时消息不丢失处理，导致整个灰度方案都是从入门到放弃的过程：
-
因为使用不同的消费组，那么灰度版本验证通过后，如何正确地衔接回原正常版本的消费组的消费位移，做到高效地不丢失信息处理呢？
-
灰度的消息如何确保准确地消费完毕，做到落在灰度标识的消息做到高效地不丢失信息处理呢？
-
开启灰度时，灰度消息的位点从那里开始？状态的细节化如何管控？
四、鲁班MQ平台的灰度方案
本质上，MQ灰度问题的核心就是高效地将灰度与非灰度的消息隔离开，消费方按照自己的需求来准确获取到对应版本的消息；当灰度完成后，能够正确地拼接回来消息的位移，做到不丢失处理必要的消息，也就是状态细节上的管理。为了实现这个目的，本方案分别在以下几点进行了改造。
本方案中涉及到的代码为测试代码，主要用于说明方案，实际代码会更精细处理。
4.1 Queue的隔离使用
![图片]()
（图4.1 Queue的区分使用）
我们已经知道了Queue是topic的实际执行单元，我们的思路就是借助Queue来实现v1（正常）消息、v2（灰度）消息的区分，我们可以固定首尾两条【可配置】Queue专门用来发送与接收灰度的消息，其余的Queue用来发送正常的线上消息。我们使用相同的消费组（也就是和业界的通用方案不一样，我们会使用相同的GroupID)，让灰度消费者参与灰度Queue的重平衡，非灰度消费者参与非灰度Queue的重平衡。
这里我们解决了消息的存储隔离问题。
4.2 Broker订阅关系改造
灰度版本往往需要变更Topic或Tag，由于我们没有新增独立的灰度消费组，当灰度版本变更Topic/Tag时，消费组内订阅关系就会不一致，前文也简单解释了订阅关系一致性的原理，我们需要在Broker做出对应的改造，来兼容灰度与非灰度订阅关系不一致的情况。
同一消费组的订阅信息会在维护在ConsumerGroupInfo的subscriptionTable中，可以在ConsumerGroupInfo中增加创建一份graySubscriptionTable用来存储灰度版本的订阅信息，客户端向Broker发送的心跳包会改造成带有自身的灰度标记grayFlag，根据灰度标记grayFlag来选择订阅关系存储在subscriptionTable还是graySubscriptionTable；在拉取消息时，同样向Broker传来grayFlag来选择从subscriptionTable还是graySubscriptionTable中获取对应的订阅信息。
这里我们解决了消费订阅一致性问题。
4.3 Producer的改造
发送方的改造相对简洁，只需确定发送的消息是否为灰度消息，通过实现MessageQueueSelector接口，将灰度消息投递到指定数量的灰度Queue即可。这里我们把用于灰度的grayQueueSize定义到配置中心中去，当前更多是约定使用Broker的指定Queue号作为灰度使用。
TOPIC_V_ORDER共有6条Queue，如图4.2所示，灰度消息只会发送至首尾的0号与5号Queue，非灰度消息则会选择其余的4条Queue发送消息。
![图片]()
![图片]()
（图4.2 发送结果）
这里我们解决了生产者正确投递的问题。
4.4 Consumer的改造
消费方涉及的改造点主要是灰度Queue与非灰度Queue的重平衡分配策略，与各个客户端灰度标记grayFlag的更新与同步。
灰度重平衡策略的核心就是分类处理灰度和非灰度的Queue，要将灰度的Queue分配至灰度ClientID，将非灰度的Queue分配至非灰度的ClientID，因此，在重平衡之前，会通过Namesrv获取同组内的所有客户端clientId对应最新的grayFlag（也就是状态会记录到Namesrv）。
当灰度版本需要变更为线上版本时，各客户端会同步grayFlag到Namesrv，同时，为了避免灰度消息还未消费完成，在更新grayFlag之前会先判断灰度Queue中是否存在未消费的消息，在保证灰度消息消费完成后才会进行grayFlag的更新。
消费者需使用AllocateMessageQueueGray作为重平衡策略，传入灰度Queue的数量，灰度消费者setGrayFlag为true，可以看出只消费了首尾的0号与5号Queue的消息，非灰度消费者setGrayFlag为false，可以看出只会消费中间的4条Queue的消息，在控制台也可以非常清晰的看到Queue的分配结果，grayFlag为true的v2客户端分配到了首尾的Queue，grayFlag为false的v1客户端则分配到了中间的4条Queue。
![图片]()
![图片]()
![图片]()
（图4.3 消费与订阅结果）
当灰度版本需要切换至线上版本时，只需调用updateClientGrayFlag来更新状态即可，可以看出在调用updateClientGrayFlag后，原先v2的两个灰度客户端在消费完灰度Queue的消息后，grayFlag才真正变为false【状态在namesrv保存】，加入到中间的4条非灰度Queue的重平衡中，原先首尾的2条灰度Queue则没有消费者订阅。
![图片]()
![图片]()
![图片]()
（图4.4 grayFlag更新）
这里我们解决了状态切换的细节控制处理问题。
4.5 Namesrv的改造
前文提到过，消费者在重平衡时是需要获取组内所有客户端的灰度标识grayFlag，因此，我们需要一个地方来持久化存储这些grayFlag，这个地方是每个消费者都可以访问的，我们选择将这些信息存储在Namesrv。
但是Namesrv本身是无状态的节点，节点之间是不会进行信息同步的，灰度数据的一致性需要借助数据库来保证，Namesrv共同访问同一套数据库就好了，数据库持久化存储灰度信息，每次更新v1、v2的灰度状态时，通过Namesrv修改数据库的数据，在每次重平衡之前，再通过Namesrv拉取自己消费组内的所有实例的灰度状态即可。
![图片]()
（图4.5 Namesrv存储灰度数据示意图）
这里我们解决了状态存储与同步的问题。
五、灰度场景的校验
测试是校验方案可行性的真理，下面用简单的demo来验证鲁班平台的MQ灰度方案。
5.1 灰度版本Topic & Tag不变
这种场景在4.3、4.4时已经做了验证，不再赘述。
5.2 灰度版本Topic增加
假设v1、v2的订阅信息如表5.1所示，则Topic订阅结果如图5.1所示，TOPIC_V_ORDER被v1、v2同时订阅，首尾两条Queue分配给灰度v2的客户端，中间4条Queue则分配给非灰度v1的客户端；TOPIC_V_PAYMENT只被灰度版本v2订阅，则只会将首尾两条Queue分配给v2的客户端，其余四条Queue不会被客户端订阅。我们向TOPIC_V_ORDER分别发送4条非灰度消息和灰度消息，向TOPIC_V_PAYMENT发送4条灰度消息，从图5.2中可以看出TOPIC_V_ORDER中的非灰度消息由v1的两个客户端成功消费，TOPIC_V_ORDER与TOPIC_V_PAYMENT的灰度消息则由v2的两个客户端成功消费。
![图片]()
（表5.1 订阅信息表）
![图片]()
（图5.1 订阅结果）
![图片]()
（图5.2 消费结果）
5.3 灰度版本Topic减少
假设v1、v2的订阅信息如表5.2所示，则Topic订阅结果如图5.3所示，TOPIC_V_ORDER被v1、v2同时订阅，首尾两条Queue分配给灰度v2的客户端，中间4条Queue则分配给非灰度v1的客户端；TOPIC_V_PAYMENT只被非灰度版本v1订阅，则只会将中间的四条Queue分配给v1的客户端，首尾两条Queue不会被客户端订阅。我们向TOPIC_V_ORDER分别发送4条非灰度消息和灰度消息，向TOPIC_V_PAYMENT发送4条非灰度消息，从图5.4中可以看出TOPIC_V_ORDER与TOPIC_V_PAYMENT的非灰度消息由v1的两个客户端成功消费，TOPIC_V_ORDER中的灰度消息则由v2的两个客户端成功消费。
![图片]()
（表5.2 订阅信息表）
![图片]()
（图5.3 订阅结果）
![图片]()
（图5.4 消费结果）
5.4 灰度版本Tag变化
假设v1、v2的订阅信息如表5.3所示，则Topic订阅结果如图5.5所示，TOPIC_V_ORDER被v1、v2同时订阅，首尾两条Queue分配给灰度v2的客户端，中间4条Queue则分配给非灰度v1的客户端，我们向TOPIC_V_ORDER分别发送4条Tag=v1的非灰度消息和Tag=v2的灰度消息，从图5.6中可以看出Tag为v1的非灰度消息由v1的两个客户端成功消费，Tag为v2的灰度消息则由v2的两个客户端成功消费。
![图片]()
（表5.3 订阅信息表）
![图片]()
（图5.5 订阅结果）
![图片]()
（图5.6 消费结果）
5.5 灰度版本Topic & Tag混合变化
假设v1、v2的订阅信息如表5.4所示，则Topic订阅结果如图5.7所示，与5.2情况相同不再赘述。我们向TOPIC_V_ORDER分别发送4条Tag=v1的非灰度消息和Tag=v2的灰度消息，向TOPIC_V_PAYMENT发送4条灰度消息，消费结果如图5.8所示，可以看出v2的两个客户端成功消费了TOPIC_V_PAYMENT及TOPIC_V_ORDER中Tag=v2的灰度消息，而v1的两个客户端则只消费了TOPIC_V_ORDER中Tag=v1的非灰度消息。
![图片]()
（表5.4 订阅信息表)
![图片]()
（图5.7 订阅结果）
![图片]()
（图5.8 消费结果）
六、结语
实际的MQ灰度版本，我们还对MQ的发送与消费方做了统一的封装，业务方只需配置graySwitch、grayFlag即可，graySwtich标记是否需要开启灰度消息，在graySwitch开启的前提下，grayFlag才会生效，用来标记当前客户端是否为灰度客户端。
在多系统交互时，业务系统可通过开关graySwitch来控制是否全量消费其他系统的灰度与非灰度消息，通过grayFlag来控制是单独消费灰度消息还是非灰度消息。graySwitch、grayFlag参数可放在配置中心做到热生效，当需要切换灰度流量时，可开发相应的脚本统一化更改grayFlag，实现全链路灰度流量的无损切换。
另外，我们对于切换状态借助Namesrv做了充分细节上的控制，保证在真正执行切换前，未消费完的消息会被消费完毕才真正的执行切换。
在此，也非常感谢阿里开源RocketMQ这个消息中间件！
作者：vivo流程IT团队-Ou Erli、Xiong Huanxin
