作者：vivo 互联网中间件团队 - Liu Tao

在开源 RocketMQ 基础之上，关于【测试环境项目多版本隔离】业务诉求的落地与实践。

一、背景

在2022年8月份 vivo 互联网中间件团队完成了互联网在线业务的MQ引擎升级，从RabbitMQ 到 RocketMQ 的平滑升级替换。

在业务使用消息中间件的过程中，提出了开发测试环境项目多版本隔离的诉求。本文将介绍我们基于 RocketMQ 如何实现的多版本环境隔离。

二、消息中间件平台主体架构

在正式展开项目多版本实践之前，先大致介绍下我们消息中间件平台的主体架构。

由上图可知，我们消息中间件平台的核心组件 mq-meta、RabbitMQ-SDK、mq-proxy，以及RocketMQ集群。

1. mq-meta

主要负责平台元数据管理，以及业务SDK启动时的鉴权寻址操作。

业务进行topic申请时，会自动分配创建到两个不同机房的broker上。

鉴权寻址时会根据业务接入Key找到所在 MQ 集群下的proxy节点列表，经过机房优先+分片选取+负载均衡等策略，下发业务对应的proxy节点列表。

2. RabbitMQ-SDK

目前业务使用的消息中间件SDK仍为原有自研的RabibitMQ SDK，通过AMQP协议收发消息。

与proxy之间的生产消费连接，遵循机房优先原则，同时亦可以人为指定优先机房策略。

3. mq-proxy

消息网关组件，负责AMQP协议与RocketMQ Remoting协议之间的相互转换，对于业务侧目前仅开放了AMQP协议。

具备读写分离能力，可配置只代理生产、只代理消费、代理生产消费这三种角色。

与broker之间的生产消费，遵循机房优先原则。

机房优先的实现：

• 生产：proxy优先将消息发送到自己本机房的broker，只有在发送失败降级时，才会将消息发送到其他机房broker；通过扩展MQFaultStrategy+LatencyFaultTolerance，并结合快手负载均衡组件simple-failover-java实现机房优先+机房级别容灾的负载均衡策略。

• 消费：在进行队列分配时，先轮询分配自己机房的队列；再将不存在任何消费的机房队列，进行轮询分配。通过扩展AllocateMessageQueueStrategy实现。

4. RocketMQ集群

每个MQ集群会由多个机房的broker组成。

每个topic则至少会分配到两个不同机房的broker上。实现业务消息发送与消费的机房级别的容灾。

每个broker部署两节点，采用主从架构部署，并基于zookeeper实现了一套自动主从切换的高可用机制。通过异步刷盘+同步双写来保证性能与消息的可靠性。

namesrv则为跨机房broker+mq-proxy之间的公共组件，为集群提供路由发现功能。

三、项目多版本实践

3.1 现状

后端服务通常采用微服务架构，各服务之间的通信，通常是同步与异步两种调用场景。其中同步是通过RPC调用完成，而异步则是通过MQ（RocketMQ）生产消费消息实现。

在多版本环境隔离中，同步调用场景，一些RPC框架都能有比较好的支持（如Dubbo的标签路由）；但在异步调用场景，RocketMQ并不具备完整的版本隔离方案，需要通过组合一些功能自行实现。

最初消息中间件平台支持的多版本环境隔离大致如下：

• 平台提供固定几个MQ逻辑集群（测试01、测试02、测试03...）来支持版本隔离。

• 业务在进行多版本的并行测试时，需关注版本环境与MQ逻辑集群的对应关系，一个版本对应到一个MQ逻辑集群。

• 不同MQ逻辑集群下用到的MQ资源（Topic、Group）自然就是不同的。

该方式主要存在如下两个问题：

1、使用成本较高

• 业务需在消息中间件平台进行多套环境（集群）的资源申请。

• 业务在部署多版本时，每个版本服务都需要配置一份不同的MQ资源接入Key，配置过程繁琐且容易出错。

2、环境维护成本较高

• 在一个项目中，业务为了测试完整的业务流程，可能会涉及到多个生产方、消费方服务。尽管在某次版本中只改动了生产方服务，但仍需要在版本环境中一并部署业务流程所需的生产与消费方服务，增加了机器与人力资源成本。

为解决上述问题，提升多版本开发测试过程中的研发效率，中间件团队开始了RocketMQ多版本环境隔离方案的调研。

3.2 方案调研

注释：

1、物理隔离：即机器层面的隔离，MQ的物理隔离，则意味着使用完全不同的MQ物理集群。

2、资源逻辑隔离：属于同一MQ物理集群，但采用不同的逻辑集群，业务侧需关注不同逻辑集群下相应的topic和group资源配置。

3、基线版本：通常为当前线上环境的版本或者是当前的主开发版本，为稳定版本。

4、项目版本：即项目并行开发中的多版本，非基线版本。

5、消息回落：针对消费而言，若消费方没有对应的项目版本，则会回落到基线版本来进行消费。

3.3 方案选择

基于我们需解决的问题，并对实现成本与业务使用成本的综合考量，我们仅考虑【基于消息维度的user-property】与【基于topic的messageQueue】这两种方案。

又因在全链路的多版本环境隔离的需求中，业务使用的版本环境明确提出不做固定，故而我们最终选择【基于消息维度的user-property】来作为我们多版本环境隔离的方案。

3.4 项目多版本的落地

基于消息维度的user-property来实现项目多版本的隔离。

1. 链路分析

在多版本环境中，真实的业务链路可能如下，服务调用可能走同步RPC或异步MQ。

注释：

1、业务请求中带有流量标识，经过网关时，根据流量路由规则将流量染色为全链路染色标识v-traffic-lane。

2、流量标识为userId，流量路由规则为用户路由到指定版本，图中的链路情况：

3、在后续的整个链路中，都需要将请求按照流量染色标识v-traffic-lane正确路由到对应版本环境。

2. 染色标识传递

为了正确识别当前服务所在版本，以及流量中的染色标识进行全链路传递，需要做如下事情：

（1）启动

其中v-traffic-lane则是服务被拉起时所在的版本环境标识（由CICD提供），这样proxy就能知道这个客户端连接属于哪个版本。

（2）消息的发送与接收

消息发送：mq-proxy将AMQP消息转化为RocketMQ消息时，将染色标识添加到RocketMQ消息的user-property中。

消息接收：mq-proxy将RocketMQ消息转化为AMQP消息时，将染色标识再添加到AMQP消息属性中。

注释：

上述红色点位，可通过改动SDK进行染色标识的传递，但这样就需要业务升级SDK了。这里我们是借助调用链agent来统一实现。

3.生产消费逻辑

（1）生产

逻辑比较简单，对于存在版本tag的消息，只需要将版本标识作为一个消息属性，存储到当前topic中即可。

（2）消费

这里其实是有两个问题：消费的多版本隔离、消息回落。

我们先看下消费的多版本隔离应该如何实现？

通过使用不同的消费group，采用基于user-property的消息过滤机制来实现。

① 版本tag传递

• 在RabbitMQ-SDK消费启动时，通过全链路Agent传递到proxy

② 项目环境消费【消费属于自己版本的消息】

• proxy会根据版本tag在MQ集群自动创建带版本tag的group，并通过消费订阅的消息属性过滤机制，只消费自己版本的消息。

• routingKey的过滤则依赖proxy侧的过滤来完成。相对基线版本，多版本的消息量应该会比较少，全量拉取到proxy来做过滤，影响可控。

• 消费组group_版本tag无需业务申请，由客户端启动时proxy会自动创建。

③ 基线消费【消费全部基线版本消息+不在线多版本的消息】

• 启动时使用原始group，订阅消费时，基于broker的routingKey过滤机制消费topic所有消息。

• 当消息被拉取到proxy后，再做一次消息属性过滤，将多版本进行选择性过滤，让基线消费到正确版本的消息。

我们再来看下消息回落又该如何实现？

1、消息回落是基线消费需要根据多版本的在线情况，来决定是否需要消费多版本的消息。

2、上面已提到基线消费从broker是拉取所有消息进行消费。

3、我们通过在基线消费内部维护一个在线多版本tag的集合，然后进行多版本消息的选择性过滤来支持回落。

4、但这个在线多版本tag的集合，需要及时更新，才能更好的保证消息回落的准确性。

5、起初我们采用定时任务从broker拉取所有在线多版本tag的集合，每30s拉取一次，这样消息回落就需要30s才能生效，准确性差。

6、后面我们想到用广播通知机制，在多版本上下线时广播通知到所有的基线消费实例，保证了消息回落的实效性与准确性。

7、完整的基线消费实例在线多版本tag集合更新机制如下：

（3）broker侧的调整

这里主要是为了配合消费多版本的实现，对broker进行了一些扩展。

1、提供在线多版本group集合的扩展接口。用以返回当前group所有在线的多版本group集合。

2、增加broker侧多版本消息过滤机制。因RocketMQ原生sql92过滤表达式，无法支持带点的属性字段过滤；而我们的版本标识（_vh_.v-traffic-lane）是存在的。

注释：

1、routingKey过滤机制：为基于broker的消息过滤机制的扩展，可实现RabbitMQ中的routingKey表达式相同的消息路由功能。

2、多版本生产消费逻辑，都在mq-proxy与RocketMQ-broker侧完成。业务也无需升级SDK。

4. 问题定位

在多版本隔离中，平台对用户屏蔽了复杂的实现细节，但用户使用时，也需要能观测到消息的生产消费情况，便于问题跟踪定位。

这里我们主要提供了如下功能：

① 消息查询：可观测消息当前的版本标识，以及消息轨迹中的生产消费情况

 

② 消费group的在线节点：可看到消费节点当前的版本标识

四、总结与展望

本文概述了vivo互联网中间件团队，在开源RocketMQ基础之上，如何落地【测试环境项目多版本隔离】的业务诉求。其中涵盖了vivo消息中间件主体架构现状、业内较流行的几种方案对比，并对我们最终选择方案在实现层面进行了细节性的分析。希望可以给业界提供一种基于proxy来实现多版本隔离特性的案例参考。

在实现过程中遇到的问题点归结下来则是：

1. 流量染色标识在整个生产消费过程中如何传递？

• 在客户端SDK使用全链路agent进行流量染色标识的添加、拆解、传递。

• 在RocketMQ则存储到消息的user-property当中。

2. 消费客户端版本标识如何识别？

• 客户端SDK使用全链路agent将版本标识添加到连接属性当中。

• proxy则根据客户端版本标识自动创建多版本消费group。

3. 消费的多版本隔离如何实现？

• 项目版本，通过不同的消费group，基于broker端消息属性的版本过滤来实现隔离。

• 基线版本，则通过proxy侧消费过滤来忽略掉不需要消费的消息。

4. 消息回落如何实现？如何保证消息回落的实效性与准确性？

• 基线版本内部会维护一个在线多版本消费group的集合，根据这个集合来决定消息是否需要回落到基线进行消费。

• 消息回落的实效性与准确性则通过定时+广播消息的机制保证。

最后，我们实现的多版本隔离特性如下：

• 多版本环境隔离。在proxy层面基于消息维度user-property来实现版本隔离，业务不需要升级SDK，业务使用层面仍然为同一套配置资源。

• 支持消息回落。

• 消费失败产生的重试消息也能被重投递到对应版本。

但仍存在如下不足：

多版本消费客户端全部下线场景：若topic中仍存在一些已下线版本的消息没有消费，则这部分消息不保证一定能被基线版本全部消费到。因基线版本与项目版本实际上采用的是不同的消费group，在broker的消费进度是不一致的，消息回落到基线消费之后，其消费位点可能已经超过项目版本消费group下线时的位点，中间存在偏差，会导致这部分消息再无法被基线版本消费到。

建议用于开发测试环境，因其无法保证多版本消息至少会被消费一次。

未来，消息中间件也会考虑线上环境全链路灰度场景的支持。

 

附录：

1. RocketMQ 全链路灰度探索与实践 + 配置消息灰度

2. 快手 RocketMQ 高性能实践 + simple-failover-java

3. 平安银行在开源技术选型上的思考和实践

4. vivo 鲁班平台 RocketMQ 消息灰度方案

5. OpenSergo

6. 从RabbitMQ平滑迁移到RocketMQ技术实战