背景

目前，公司方面 RPC 调用如 Dubbo、Feign 已经能支持基于灰度的调用，但是 MQ 还没有支持灰度的能力，因此导致在测试和生产环境业务验证、消息隔离方面体验比较差，因此我们基于 RabbitMQ 和 Kafka 实现了消息灰度的能力。

灰度场景

大部分场景下 MQ 的灰度并不会像 RPC 那样那么严格，但是我们需要确认消费场景，即当灰度消费者不存在的情况下，消息是否应该由正常消费者去消费。

1. 灰度消息只由灰度节点消费

事实的情况是可能大家都想要这种严格意义上的消息灰度隔离策略，由此才证明是真正的消息灰度方案，但是这个方案需要考虑一些具体场景问题。

比如，有时候作为灰度节点的发送方，它的功能改动点并不是在 MQ 这块，但是它发送的消息却是灰度消息，而消息的消费方可能也未发生过功能变动，也不会有与之对应的灰度消费标识，这种情况下如果我们将灰度的消息进行丢弃的话，那么会造成最终的数据不完整。

2. 灰度消息可以由正常节点消费

因此，我们再考虑第二种方案，如果当灰度消费节点不存在时，消息会由正常节点消费，当存在灰度节点时，则由灰度节点消费，正常节点消费灰度消息只为了当灰度节点不存在时的兜底。

那么，这种场景仍然可能存在问题，比如当消费节点的消费逻辑发生改变时，由正常节点消费就可能造成业务上的错误。对于此问题我们可以默认认为如果消费方发生逻辑改变，则灰度节点大概率一定是存在的，如果一些异常情况导致的异常或者宕机的场景，仍然能通过人工或者告警判断出来，总之，这个问题认为不算是问题。

灰度方案

我们分别从 MQ 的自身特性和一些通用的处理方式出发，分别探讨 RabbitMQ 和 Kafka 的灰度实现方式。

常规方案：影子Queue/Topic

这个是现在实现 MQ 灰度最为常见的方案，为每一个Queue/Topic都建立一个与之对应的灰度Queue/Topic。

生产者层将要发送的消息进行Queue/Topic/RoutingKey的动态修改，让他发送到灰度或正常的Queue/Topic中。

而消费者层面只需要在应用启动时根据自身的灰度标记动态的切换到灰度Queue/Topic进行监听即可。

但是对于我们目前的系统现状而言，这个方案存在三个问题：

首先，由于我们目前系统测试环境的灰度标签是可以定制的，可能每一个功能上线都会有一个对应的灰度标识，这样带来的问题就是Queue/Topic的数量会随着灰度标识的增加而倍数性的增加。

而不管哪种MQ，过多的Queue/Topic都会对 MQ 本身造成一定处理能力下降。

另外，我们的灰度标签是可以根据启动的实例随意修改的，也就意味着对应的整套Queue/Topic也得跟着灰度的标识随意的创建。这样一来，人工手动跟着创建显然就不太现实，而生产环境中我们的Queue/Topic创建是需要走流程申请的，这又和我们的现状违背。

再者，即便我们能够根据生产者的灰度标识动态的创建Queue/Topic的话，那么至少也需要考虑在灰度生产者实例正常下线时将它创建的Queue/Topic进行销毁，如果异常的下线还需要人工的接入定期的进行Queue/Topic的清理工作。

最后，如果是针对 Kafka 或 RocketMQ，这种方案实行起来还比较简单，如果是对于RabbitMQ，这里又多了一层 Exchange 和 Queue 的绑定关系，不同的生产模式也需要去做各自的适配。

所以，为了在 RabbitMQ 和 Kafka 之间的一致性，我们决定不采用该方案来实现。

RabbitMQ

对于 RabbitMq，我们使用重新入队这个特性来实现灰度队列。

通过重新入队的这个特性，我们可以在生产者发送消息时将灰度的标识标记到消息头，发送时一并发出。

当消费者消费消息时，根据消费者自身标记决定要不要对消息进行消费，如果消费者本身不满足灰度消费规则，则把这条灰度消息进行Requeue处理。

这条消息经过轮询，最终会流转到灰度标识的消费者进行消费。

Requeue

实现思路生产者在发送消息之前获取到当前实例的灰度标记，对消息 Header 添加灰度标记对消费者添加监听器，灰度节点消费根据灰度标记判断对灰度消息的消费，正常节点根据开关决定是否消费或者进行 Requeue生产流程

生产者在启动时，我们通过自动装配，注册 RabbitTemplate 时setBeforePublishPostProcessors添加前置处理器，在发送消息前对消息的 Header 添加灰度标记。

消费流程

首先，在消费时通过监听SimpleMessageListenerContainer重写executeListener方法进行消息处理。

当灰度开关未打开，执行正常消费逻辑。当灰度机器直接匹配到灰度消息时，那么直接消费即可。通过监听 Eureka 本地缓存刷新的事件不停地刷新灰度实例的缓存，当正常节点消费灰度消息时，如果灰度实例不存在就可以直接消费。如果存在灰度实例且正常节点消费到灰度消息，考虑两种可能，第一是正常的轮询到正常节点，第二是灰度节点prefetch_count达到阈值，阻塞队列已满，灰度消息在正常节点之间不停地轮询。为了解决第二个场景，添加了一层布隆过滤器，当再次匹配到同样的消息时，当前节点将休眠一段短暂的时间。上述场景都未匹配到，那么执行 Requeue 操作。Kafka

在 Kafka 的消费理念中有一层消费者组的概念，每个消费者都有一个对应的消费组。

当消息发布到主题后，只会被投递给订阅它的每个消费组中的一个消费者，两个消费组之间互不影响。

借助这个消费特性，可以将同一个消费组中的灰度消费者单独拎出来，做成一个特殊的消费组，这样每个消费组都会接收到同样的消息。

在正常的消费组中，遇到带有灰度标识的消息，我们只做空消费，不实际执行业务逻辑，在灰度消费组中的消费者，只处理匹配到灰度规则的消息，其它的消息做空消费。

实现思路生产者生产灰度消息的时候在消息 Header 里面添加灰度标记灰度消费者和正常消费者设置不同的GroupId灰度消费者和正常消费者在拿到消息后判断有没有灰度标记，判断配置中心是否开启了消息灰度，如果开启了则进行灰度节点的消费，如果没开启则不消费生产流程

生产者在启动的时候会去动态装配所有的拦截器，装配的方式为在 BeanPostProcessor 的后置处理器中获取到 KafkaTemplate 对象，把我们的拦截器的类的全限定名 set 进去 config 即可，这里可以支持不管用户自己创建的 Factory对象还是 KafkaTemplate 对象都能进行拦截器的装配。

消费流程

消费的时候也是一样，如果当前节点是灰度节点，那么就修改当前group.id为灰度，最后通过拦截器执行消费逻辑。