Apache RocketMQ 的 Service Mesh 开源之旅

做者 | 凌楚   阿里巴巴开发工程师

导读：自 19 年末开始，支持 Apache RocketMQ 的 Network Filter 历时 4 个月的 Code Review（Pull Request），于本月正式合入 CNCF Envoy 官方社区（RocketMQ Proxy Filter 官方文档），这使得 RocketMQ 成为继 Dubbo 以后，国内第二个成功进入 Service Mesh 官方社区的中间件产品。

Service Mesh 下的消息收发

主要流程以下图：

图 1

简述一下 Service Mesh 下 RocketMQ 消息的发送与消费过程：

• Pilot 获取到 Topic 的路由信息并经过 xDS 的形式下发给数据平面/Envoy ，Envoy 会代理 SDK 向 Broker/Nameserver 发送的全部的网络请求；
• 发送时，Envoy 经过 request code 判断出请求为发送，并根据 topic 和 request code 选出对应的 CDS，而后经过 Envoy 提供的负载均衡策略选出对应的 Broker 并发送，这里会使用数据平面的 subset 机制来确保选出的 Broker 是可写的；
• 消费时，Envoy 经过 request code 判断出请求为消费，并根据 topic 和 request code 选出对应的 CDS，而后和发送同样选出对应的 Broker 进行消费（与发送相似，这里也会使用 subset 来确保选出的 Broker 是可读的），并记录相应的元数据，当消息消费 SDK 发出 ACK 请求时会取出相应的元数据信息进行比对，再经过路由来准确将 ACK 请求发往上次消费时所使用的 Broker。

RocketMQ Mesh 化所遭遇的难题

Service Mesh 经常被称为下一代微服务，这一方面揭示了在早期 Mesh 化浪潮中微服务是绝对的主力军，另外一方面，微服务的 Mesh 化也相对更加便利，而随着消息队列和一些数据库产品也逐渐走向 Service Mesh，各个产品在这个过程当中也会有各自的问题亟需解决，RocketMQ 也没有例外。

有状态的网络模型

RocketMQ 的网络模型比 RPC 更加复杂，是一套有状态的网络交互，这主要体如今两点：

• RocketMQ 目前的网络调用高度依赖于有状态的 IP；
• 原生 SDK 中消费时的负载均衡使得每一个消费者的状态不能够被忽略。

对于前者，使得现有的 SDK 彻底没法使用分区顺序消息，由于发送请求和消费请求 RPC 的内容中并不包含 IP/(BrokerName + BrokerId) 等信息，致使使用了 Mesh 以后的 SDK 不能保证发送和消费的 Queue 在同一台 Broker 上，即 Broker 信息自己在 Mesh 化的过程当中被抹除了。固然这一点，对于只有一台 Broker 的全局顺序消息而言是不存在的，由于数据平面在负载均衡的时候并无其余 Broker 的选择，所以在路由层面上，全局顺序消息是不存在问题的。

对于后者，RocketMQ 的 Pull/Push Consumer 中 Queue 是负载均衡的基本单位，原生的 Consumer 中实际上是要感知与本身处于同一 ConsumerGroup 下消费同一 Topic 的 Consumer 数目的，每一个 Consumer 根据本身的位置来选择相应的 Queue 来进行消费，这些 Queue 在一个 Topic-ConsumerGroup 映射下是被每一个 Consumer 独占的，而这一点在现有的数据平面是很难实现的，并且，现有数据平面的负载均衡无法作到 Queue 粒度，这使得 RocketMQ 中的负载均衡策略已经再也不适用于 Service Mesh 体系下。

此时咱们将目光投向了 RocketMQ 为支持 HTTP 而开发的 Pop 消费接口，在 Pop 接口下，每一个 Queue 能够再也不是被当前 Topic-ConsumerGroup 的 Consumer 独占的，不一样的消费者能够同时消费一个 Queue 里的数据，这为咱们使用 Envoy 中原生的负载均衡策略提供了可能。

图 2

图 2 右侧即为 Service Mesh 中 Pop Consumer 的消费状况，在 Envoy 中咱们会忽略掉 SDK 传来的 Queue 信息。

弹内海量的 Topic 路由信息

在集团内部，Nameserver 中保存着上 GB 的 Topic 路由信息，在 Mesh 中，咱们将这部分抽象成 CDS，这使得对于没法预先知道应用所使用的 Topic 的情形而言，控制平面只能全量推送 CDS，这无疑会给控制平面带来巨大的稳定性压力。

在 Envoy 更早期，是彻底的全量推送，在数据平面刚启动时，控制平面会下发全量的 xDS 信息，以后控制平面则能够主动控制数据的下发频率，可是无疑下发的数据依旧是全量的。后续 Envoy 支持了部分的 delta xDS API，便可如下发增量的 xDS 数据给数据平面，这固然使得对于已有的 sidecar，新下发的数据量大大下降，可是 sidecar 中拥有的 xDS 数据依然是全量的，对应到 RocketMQ ，即全量的 CDS 信息都放在内存中，这是咱们不可接受的。因而咱们但愿可以有 on-demand CDS 的方式使得 sidecar 能够仅仅获取本身想要的 CDS 。而此时正好 Envoy 支持了 delta CDS，并仅支持了这一种 delta xDS。其实此时拥有 delta CDS 的 xDS 协议自己已经提供了 on-demand CDS 的能力，可是不管是控制平面仍是数据平面并无暴露这种能力，因而在这里对 Envoy 进行了修改并暴露了相关接口使得数据平面能够主动向控制平面发起对指定 CDS 的请求，并基于 delta gRPC 的方式实现了一个简单的控制平面。Envoy 会主动发起对指定 CDS 资源的请求，并提供了相应的回调接口供资源返回时进行调用。

对于 on-demand CDS 的叙述对应到 RocketMQ 的流程中是这样的，当 GetTopicRoute 或者 SendMessage 的请求到达 Envoy 时，Envoy 会 hang 住这个流程并发起向控制平面中相应 CDS 资源的请求并直到资源返回后重启这个流程。

关于 on-demand CDS 的修改，以前还向社区发起了 Pull Request ，如今看来当时的想法仍是太不成熟了。缘由是咱们这样的作法彻底忽略了 RDS 的存在，而将 CDS 和 Topic 实现了强绑定，甚至名称也如出一辙，关于这一点，社区的 Senior Maintainer [@htuch ]() 对咱们的想法进行了反驳，大意就是实际上的 CDS 资源名可能带上了负载均衡方式，inbound/outbound 等各类 prefix 和 suffix，不能直接等同于 Topic 名，更重要的是社区赋予 CDS 自己的定义是脱离于业务的，而咱们这样的作法过于 tricky ，是与社区的初衷背道而驰的。

所以咱们就须要加上 RDS 来进行抽象，RDS 经过 topic 和其余信息来定位到具体所须要的 CDS 名，因为做为数据平面，没法预先在代码层面就知道所须要找的 CDS 名，那么如此一来，经过 CDS 名来作 on-demand CDS 就更无从谈起了，所以从这一点出发只能接受全量方案，不过好在这并不会影响代码贡献给社区。

route_config:
  name: default_route
  routes:
  - match:
      topic:
        exact: mesh
      headers:
        - name: code
          exact_match: 105
    route:
      cluster: foo-v145-acme-tau-beta-lambda

上面能够看到对于 topic 名为 mesh 的请求会被 RDS 路由到 foo-v145-acme-tau-beta-lambda 这个 CDS 上，事先咱们只知道 topic 名，没法知道被匹配到的 CDS 资源名。

现在站在更高的视角，发现这个错误很简单，可是其实这个问题咱们直到后续 code review 时才及时纠正，确实能够更早就作得更好。

不过从目前社区的动态来看，on-demand xDS 或许已是一个 roadmap，起码目前 xDS 已经全系支持 delta ，VHDS 更是首度支持了 on-demand 的特性。

Mesh 为 RocketMQ 带来了什么？

A service mesh is a dedicated infrastructure layer for handling service-to-service communication. It’s responsible for the reliable delivery of requests through the complex topology of services that comprise a modern, cloud native application. In practice, the service mesh is typically implemented as an array of lightweight network proxies that are deployed alongside application code, without the application needing to be aware.

这是 Service Mesh 这个词的创造者 William Morgan 对其作出的定义，归纳一下：做为网络代理，并对用户透明，承担做为基础设施的职责。

图 3

这里的职责在 RocketMQ 中包括服务发现、负载均衡、流量监控等职责，使得调用方和被代理方的职责大大下降了。

固然目前的 RocketMQ Filter 为了保证兼容性作出了不少让步，好比为了保证 SDK 能够成功获取到路由，将路由信息聚合包装成了 TopicRouteData 返回给了 SDK ，可是在理想状况下，SDK 自己已经不须要关心路由了，纯为 Mesh 情景设计的 SDK 是更加精简的，再也不会有消费侧 Rebalance，发送和消费的服务发现，甚至在将来像消息体压缩和 schema 校验这些功能 SDK 和 Broker 或许均可以不用再关心，来了就发送/消费，发送/消费完就走或许才是 RocketMQ Mesh 的终极形态。

图 4

What's Next ?

目前 RocketMQ Filter 具有了普通消息的发送和 Pop 消费能力，可是若是想要具有更加完整的产品形态，功能上还有一些须要补充：

• 支持 Pull 请求：如今 Envoy Proxy 只接收 Pop 类型的消费请求，以后会考虑支持普通的 Pull 类型，Envoy 会将 Pull 请求转换成 Pop 请求，从而作到让用户无感知；
• 支持全局顺序消息：目前在 Mesh 体系下，虽然全局顺序消息的路由不存在问题，可是若是多个 Consumer 同时消费全局顺序消息，其中一个消费者忽然下线致使消息没有 ACK 而会致使另外一个消费者的消息产生乱序，这一点须要在 Envoy 中进行保证；
• Broker 侧的 Proxy：对 Broker 侧的请求也进行代理和调度。

蜿蜒曲折的社区历程

起初，RocketMQ Filter 的初次 Pull Request 就包含了当前几乎所有的功能，致使了一个超过 8K 行的超大 PR，感谢@天千 在 Code Review 中所作的工做，很是专业，帮助了咱们更快地合入社区。

另外，Envoy 社区的 CI 实在太严格了，严格要求 97% 以上的单测行覆盖率，Bazel 源码级依赖，纯静态连接，自己无 cache 编译 24 逻辑核心 CPU 和 load 均打满至少半个小时才能编完，社区的各类 CI 跑完一次则少说两三个小时，多则六七个小时，并对新提交的代码有着极其严苛的语法和 format 要求，这使得在 PR 中修改一小部分代码就可能带来大量的单测变更和 format 需求，不过好的是单测能够很方便地帮助咱们发现一些内存 case 。客观上来讲，官方社区以这么高的标准来要求 contributors 确实能够很大程度上控制住代码质量，咱们在补全单测的过程当中，仍是发现并解决了很多自身的问题，总得来讲仍是有必定必要的，毕竟对于 C++ 代码而言，一旦生产环境出问题，调试和追踪起来会困可贵多。

最后，RocketMQ Filter 的代码由我和@叔田 共同完成，对于一个没什么开源经验的我来讲，为这样的热门社区贡献代码是一次很是宝贵的经历，同时也感谢叔田在此过程当中给予的帮助和建议。