作者：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技术实战