RocketMQ Streams 在云安全及 IoT 场景下的大规模最佳实践

作者：阿里巴巴云原生

2022 年 10 月 10 日
浙江
本文字数：7411 字
阅读完需：约 24 分钟

1.jpeg

本文作者：袁小栋，Apache RocketMQ Committer，RocketMQ Streams Cofonder，阿里云安全智能计算引擎负责人

RocketMQ Streams 简介

2.png

RocketMQ Streams 包含以下四个部分的定义：

（1）Lib 包：轻量，启动即可运行。只需要从 git 下载源码，编译成 jar 包即可使用。

（2）SQL 引擎：兼容了 Flink 的 SQL 语法，也兼容了 UDF、UETF 和 UDAF，可以用 Flink 的语法或将 Flink 任务直接迁移并进行使用。

（3）轻量的边缘计算引擎：RocketMQ Streams 和 RocketMQ 做了深度的集成。因为 RocketMQ 支持 MQTT，所以 RocketMQ Streams 支持云计算的场景。此外，RocketMQ 还支持消息的存储和转存，因此基本能够满足边缘计算的大部分场景。

（4）SDK：其组件可以独立使用，也可以嵌入到业务里使用。

3.png

现有的大数据框架比如 Flink、Spark、Storm 等都已经十分成熟，而我们在此基础上依然研发 RocketMQ Streams 这样一个开放框架的原因，主要基于以下考虑：

Flink 是一个底座比较重的大数据组件，集群开销和框架开销占比较大，运维成本也比较高，因此适合做中台，由专门的运维人员部署，形成一个大的中台业务。

但是实际业务中必然存在中台无法满足的场景，比如某产品依赖大数据的能力，需要将产品输出给用户，在用户的 IDC 里部署。如果将大数据计算能力也携带一起部署，则会产生三个问题：第一，部署麻烦，因为 Flink 的部署成本比较高；第二，运维成本较高；第三，资源问题，Flink 的任务需要提前预设资源，不同的用户日志量不一样，预设资源会很大，Flink 无法满足需求。

RocketMQ Streams 的定位是适合随产品输出的场景，不适合中台。比如安全风控、边缘计算、消息队列、流计算等，都适合 RocketMQ Streams。因此，RocketMQ Streams 和 Flink 的能力可以互为补充。

4.png

RocketMQ Streams 具有以下特点：

（1）轻量：RocketMQ Streams 是轻量的，1core 1g 即可部署；依赖较轻，除了消息队列没有其他依赖；发布简单，可通过 SQL 热更新的方式发布。

（2）高扩展：RocketMQ Streams 可以扩展 Source、Sink、UDF 等。

（3）高性能：对过滤做了很多优化，因此在高过滤场景下，性能可提升 3-5 倍；RocketMQ Streams 也实现了一些任务的轻量化，在 SQL 同源任务归并的场景下，资源可节省 50%；同时，它基于流计算，可以实现毫秒延迟。

（4）多部署模式：jar 包即服务；可以基于 C/S 模式通过提交 SQL 热发布，也可以通过 SDK 集成到业务里。

（5）超大维表支持：支持超大的维表，RocketMQ Streams 自研的缓存内存占比仅为 Java Map 的 16.7%；同机器上的多个任务可以共享，节省资源；维表支持千万级别，不需要指定索引，可根据 join 条件自动推断索引，做到类似 map 的 O(1)匹配。

（6）丰富功能：支持精确计算一次以及灵活的窗口，比如滚动窗口、滑动窗口、分发窗口，也支持双流 Join、维表 Join、转化、过滤、大数据开发场景等功能。

5.png

上图为 RocketMQ Streams 支持的一些常规大数据算子，与其他大数据框架基本相似，可以进行扩展。

RocketMQ Streams 架构及实现原理

6.png

无论是 Spark 还是 Flink，一个成功的大数据组件往往都需要由一个很大的团队经历几年的时间才能打磨完成。实现 RocketMQ Streams 主要会面临以下挑战：

大数据计算功能多且架构复杂，是否能够实现？
与 Flink 等大数据框架的核心差异是什么，是否会做成 Flink 的裁剪版？

7.png

实现一个轻量级、高性能的大数据计算框架，必须要有和 Flink 不一样的思路。

从业务架构分析，一个常规 RocketMQ 业务的架构基本相同，包括输入、无状态的计算、输出结果等。这种常规的 RocketMQ 业务架构有两个优点：首先，比较轻量，负载均衡、容错都由 RocketMQ 完成，不需要另外做；其次，部署简单，如果 RocketMQ 阻塞，直接扩容业务、增加消费能力即可。

但是这种常规架构很难实现统计、join 以及窗口计算等复杂计算。要实现此类复杂计算，必须实现 shuffle，而要实现 shuffle 则必须实现不同算子之间的通讯。算子之间的通讯需要有全局的调度和全局的任务管理，而全局的调度和全局的任务管理又需要资源的管理和对任务资源的分配。上述的需求会导致架构变得复杂，使短时间内的实现存在稳定性和复杂性等方面的困难。

逆向思考可以看到复杂性的根源是 shuffle，解决思路是借助消息队列的中转实现 shuffle。以 shuffle 作为分割，将复杂的拓扑变为简单的拓扑。只需重点突破整个架构的搭建、窗口计算的补充、性能的提升这三个难关，即可实现一个既轻量又有高性能的大数据计算功能框架。

8.png

大数据架构包括 Spark、Flink 等，常规设计思路是计算和集群的管理一体化。集群的管理要解决高可用问题、task 分配和调度问题、job 和 task 容错问题，因此大数据架构的实现存在巨大挑战：

（1）集群的管理需求使架构更重，因为高可用意味着必须引入组件。而且在资源消耗方面，一个集群模式至少需要三个阶段，而集群的开销可能需要 10%的内存。一旦管理结构集群化，任务的分配、资源的设定都需要预设。

（2）类似窗口计算的状态存储要求比较高。大数据组件的部署对内存、大磁盘有要求，而这种要求无疑会增加架构的复杂性。

（3）通过消息队列中转来实现 shuffle 的方案可能会加大 RocketMQ 的压力，增加部署的复杂性。

以上三点是基于大数据架构来思考实现一个轻量化架构的挑战。换一种思路，聚焦于核心业务，用业务架构的思路去思考。

常规的大数据业务都会有一个消息队列，无论消息队列是不是 RocketMQ。而大多数消息队列都会实现分片的负载均衡和容错的管理，计算和管理的分离可以借用 MQ 的集群能力，存储可以采用 RocketMQ 的压缩存储来实现。

MQ 的最小调度单元是分片，它可以对分片进行负载均衡、容错、调度等操作。只要将任务和分片进行映射，借用 MQ 的分片管理，即可实现 task 管理，无需额外实现管理能力。复用 RocketMQ 的压缩存储，也不需要额外实现存储。此外，用 MQ 做 shuffle 会加大 MQ 的压力。MQ 的消息量增加，使得 CPU 的使用率也会增加，整体资源使用率也会增加，因此要采用策略来降低资源消耗。

窗口计算的实时性不高，比如 10 分钟的窗口只需要每 10 分钟取出结果。因此可以采用微批的方式，比如 1000 条计算一次，将 1000 条基于 shuffle key 进行分组，分组以后多条数据合并成一条。RocketMQ 基于 QPS 的压力，数据量变大，QPS 下降，CPU 的压力反而不大，然后进行压缩，将数据量降低，最终可以减少 shuffle 开销。

最终结果如下：

（1）采用 shared-nothing 架构，没有任何集群和框架的开销。

（2）轻依赖：没有增加任何额外的依赖，虽然有 MQ 依赖，但 MQ 是业务必需的，可以直接复用业务的 MQ。

（3）计算机不需要任何依赖，部署轻量，1core 1g 即可部署。

（4）轻消耗：shuffle 的中转实现了微批、压缩和多条归并的策略，7000 的 QPS 只需要 0.12 的 CPU 和 300 兆的内存，资源消耗非常低。

（5）轻扩容：扩容非常简单，因为采用 shared-nothing 架构，类似于 web 服务器，消息堆积时增加实例即可扩容。

9.png

窗口的精确计算一次（Exactly-once）是一个难点。流是无边界的，要进行统计计算，则必须划分窗口。如上图，假设 D 的位置是一个 count，计算 10 分钟一共接收多少条数据，有两种实现方式：

（1）每一条数据过来都进行缓存，每十分钟将所有数据进行统计取得结果。这种方式对存储的压力比较大，也不高效。

（2）比较优雅的方式：每来一条数据都只存中间结果，比如第一条数据，中间结果是 1，第 2 条是 2，第 3 条是 3。但这也存在一个问题，如果某个节点出问题或某个任务出问题，中间结果会变为不可控的状态。假如 3 宕机， 2 和 1 可能会继续完成计算，也可能出现问题不进行计算。因此在 B 被拉起的时候回放哪条消息是不可知的，这种情况无法成为精确计算一次。

Flink 是业界精确计算最优雅的方案。它的思路很简单，在某个时间点将整个集群的状态进行一次镜像，每隔一段时间镜像一次。出现问题时，将所有算子的状态恢复一遍再计算。

整体的计算流程如下：job manager 定期发 checkpoint 给它的数据源。发生两个 checkpoint 时，checkpoint 会随着数据在算子里走。每个算子接收到 checkpoint 时，需要备份自己的状态，比如窗口算子接收到一个 checkpoint，还无法进行状态备份，需要等到另一个 checkpoint 也到了之后才能做状态的备份，该设计称为对齐等待。等待的过程取决于两个 checkpoint 之间流速的差异。等两个 checkpoint 都到了以后，再同步地进行状态存储，将本地的状态存储到远程的状态。

以上过程开销较大，打开 checkpoint 使任务性能会降低约 30%。任务越复杂，系统的开销越大。此外，恢复时长也需要考虑，当算子和任务出问题重启时，必须从远程读取完整的状态，所有算子恢复以后才能开始计算。恢复过程可能需要几秒到几分钟，时间较长。

Flink 虽然是一个优雅的方案，但依然存在很多重操作，这是因为 Flink 方案从整个拓扑考虑，因此思考点较为复杂。

而简化的思路为，将复杂的拓扑通过 shuffle 拆分成很多简单的子 Job。每个子 Job 的逻辑也很简单，包括三点：第一，从 source 接收数据；第二，进行算子的计算；第三，将数据写到 Sink。有些子 Job 算子是有状态，而有些是无状态，无状态的算例只需要保证至少消费一次的逻辑即可。

以上思路可能带来的后果是输出的 Sink 里存在重复的数据。如果该 Sink 是最终的结果，则由 Sink 自己决定能不能去重；如果是 shuffle 的队列，则会在后面有状态的算子里完成精确计算一次的逻辑。

而有状态的消费数据里存在重复数据，只需进行去重。去重的逻辑如下：在状态存储时，除了存储中间的计算结果，还需将元数据进行存储。元数据指现有的中间结果计算用到的分片以及分片的最大 offset 对应的数据。数据来的时候，如果该分片的 offset 比已经计算的小，则将它丢弃，从而通过去重完成了精确计算一次的逻辑。远程存储只需存储一份，无需定时地存储一份完整数据。

另外，Checkpoint 也不会阻塞流程，因为一个 Checkpoint 的发送只是负责获取算子当前已经存储到远程的元数据，而算子的存储过程完全可以异步和微批地进行存储。Checkpoint 到达算子后，只需要告诉其结果，不会产生任何阻塞。

消息源存储 offset 是基于所有状态算子里面的分片元素，取每个相同分片里的最小值存储，则崩溃后恢复时一定能保证至少消费一次。此处可以有重复的数据，可通过去重保证精确计算一次。

10.png

SQL 优化器是阿里云云盾的需求。因为需要将公共云的规则迁到专有云，而专有云的资源有限，只能用原先 4%的内存和不到 30%的 CPU 去运行原先 1.2 倍的规则。而专有云的另一特点是扩容成本较高，可能按月扩容，很难扩机器。

综上，公共云迁移到专有用，存在两个巨大挑战：

第一，要用有限的资源承载更多的规则；

第二，安全场景需要不停地增加规则来保证安全性，规则要增加，但需要保证资源不随规则增加。

因此，进行优化的时候也需要考虑安全的特点。而大部分大数据计算都具有此特点，所以这是一个通用的方案。

基于安全的特点来分析，安全的特点有三：

（1）正则或过滤类的表达式比较多，可以使用更快的引擎来承载。

（2）一些表达式的字段重复率较高，比如命令行，无论有多少个参数，运维工作的命令都很相似。

（3）数据源比较少，但是每个数据源的规则比较多。

基于以上三个特点，我们思考整体的解法如下：

（1）任务归并，减少任务开销。每起一个任务都会有一个线程池，占有一定的内存开销。而这些任务来自同一个数据源，因此可以将同数据源的公共部分抽取出来，比如对消费数据源实现部分字段的标准化，而对应的规则可以封装成大任务。按照以上逻辑，10 个任务放在一起，只需要 5core 5g，所以资源消耗更少，线程更少，内存使用也更少。该解法称为同源归并逻辑。

而资源变少带来的问题是会将这一组任务的容错放在一起，一个任务有问题也会影响到其他任务。因此，用户可以按需选择任务类型，分为资源敏感型和错误敏感型。

另外，同源归并不会导致规则放在一起变复杂而使开发测试变得更困难。因为对于开发测试，每一个规则依然是独立的，称为动态同源归并。同源归并是下发一个策略后自动归并，将策略撤销则会恢复为独立运行，这是可以动态调配的策略。

（2）表达式指纹。一个规则里有很多过滤条件，解决思路是将所有任务里的过滤条件都在编译期间统一收集。

一个过滤的表达式基本是三元组，包括变量、操作、值。比如正则，变量是 command line，操作是正则，值是正则的串。按变量进行分组，比如一个 command line 有 10 个表达式，另一个 command line 有 20 个表达式，放在一起是 30 个表达式，将它分为一个表达式分组。

表达式分组的目的是缓存。一条消息到达时，处理流程为：先检查缓存，所有表达式分组逐个检查。如果缓存里存在，则直接应用结果；否则，将这一组表达式全部计算完成然后生成结果。生成的结果是一个 bit set，比如 100 个表达式会有一个 100 个 bit 位，代表该表达式是否触发。

将 command line 和 bit set 放在缓存里，再来一条相同 command line 的时候，所有表达式都不需要计算，可直接获得结果。然后查看上下文是否存在该结果，如果存在则直接使用，否则再进行计算。按照该流程，如果 command line 的重复率较高，比如有 80%的重复率，只有 20%的 command line 会被真正地计算，其他的只需 O(1)时间来获取结果。

在字段重复率较高的场景中，此策略需要的计算资源大幅下降，因为它能将复杂的正则计算转化成 O(1)的比对计算，资源不会随着规则增加而增加。

此外，使用正则的场景中字段重复率比较高，这是一个通用的特性。但即使重复率比较低，因为整体资源开销只增加了一个 O(1)的比对，不会增加额外的开销。

（3）Hyperscan 正则加速。将所有任务的表达式放在一起，对表达式进行预编译，尤其是正则类。比如用 Hyperscan 可以对 1000 个表达式进行预编译，每个表达式单个执行与预编译在一起执行，两者之间约有 10 倍差距。如果字段重复率不高，可以用 Hyperscan 加速正则的执行。

流式数据量非常大，存在缓存是否能撑住以及用什么缓存来承载的问题。

首先，缓存是否能撑住可以通过设定边界解决，比如设定 300 兆的缓存，超过则丢掉一些数据。

其次，可以采用压缩缓存来承载，用很少的资源能够承载大的数据量。Java 的 map 之所以占用资源较大，是因为存在很多同步、对齐、指针开销。所以缓存只能基于原生的 bit 数组实现，可以降低资源开销。一个 key 可能比较长，比如一个 command line 可能需要几十个或几百个字符。但使用 md5 存储可以压缩到十几个字节，而且 md5 的冲突率、碰撞率非常低。

因此，使用 md5 保存 key，无论 key 多大，都能够压缩到 16 个字节。value 则用原始的字节保存，没有任何头部开销。

测试显示，50Byte 的 key，20Byte 的 value，1000 万的数据用压缩存储可以达到原始数据的一半，达到 Java map 的 17%，压缩效果显著。

最终效果：1.2 倍的安全规则，采用 32core 40g 支撑 12000QPS，没有增加任何物理机，也不需要扩容，可满足需求。

RocketMQ Streams 在阿里云安全的应用

11.png

RocketMQ Streams 的第一个应用是专有云的安全。

专有云和公有云不一样，专有云的资源有限，将整个云部署在用户的 IDC 机房，在用户机房里扩资源需要向用户申请和采购，不像公有云可以随时扩充资源。因此专有用的弹性不如公共云。

大数据计算是一个产品，不是用户购买以后才会输出。用户买了安全，但不一定买大数据计算，这种情况造成用户买了安全却没有大数据计算。如果因为安全而帮用户买大数据计算，大数据计算的成本可能比安全更高，导致了大数据落地很难，入侵检测的能力较差，风险较高。因此，我们最终的策略是采用 RocketMQ Streams 方案为用户实现部署。

RocketMQ Streams 方案需要 32core 40g 的内存，即可承载 12000QPS，基本能满足用户的需求；对比内存资源，只使用了原先公共云内存资源的 4%，CPU 的 30%不到；从能力覆盖层面看，覆盖了全部安全规则，也兼容了 Flink 的语法规则，只需开发一次；从安全效果层面看，因为覆盖了所有的安全规则，实现了安全效果 100%保障；从产品覆盖层面看，有多个产品在应用 RocketMQ Streams。

12.png

RocketMQ Streams 的第二个应用场景是云安全中心的混合云。Gartner 预测将来 80%的企业都会采用混合云和多云的部署模式，但是混合云和多云需要统一的安全运营管理。

多云或边缘计算存在的问题主要在于，比如在阿里购买了一些 ECS，在腾讯、华为购买了一些 ECS，在国外也购买了一些 ECS，如果要将 ECS 的数据汇聚在一起，日志量较大，上传成本也高。国外的 ECS 数据回到国内会受到一些限制，除此之外，如果带宽不够，上传日志也可能影响正常业务。

我们提供的解决策略很简单，将 RocketMQ Streams 和 RocketMQ 整合，支持消息队列存储，也支持 ETL 和流计算，部署到边缘端。比如阿里作为一个统一管控区，在腾讯购买两台 ECS，将 RocketMQ Streams 和边缘计算引擎部署上，可称为边缘计算。在边缘计算告警，丢掉原始日志，只回传告警。原始日志在本地还可转存给用户，如果用户需要，也能支持热更新。只需一个 zip 包，一键 SH 即可安装。

13.png

RocketMQ Streams 的第三个应用场景是 IOT。IOT 是典型的边缘计算场景，挑战在于需要用 4core 8g 来混部其业务和 RocketMQ Streams 任务，几百个任务的压力非常大。而且它需要采用 MQTT 这样标准的 IOT 协议输入，也需要自定义规则引擎的能力进行统计计算、特征计算、Join 计算和维表计算。

RocketMQ Streams 的 MQTT 直接复用了 RocketMQ。维表方面，可以实现维表共享，支持千万级维表，可在同一台机器共享多个实例。SQL 则可以支持归并和热更新。

最终在 IOT 场景中完成了 RocketMQ Streams 的能力建设和用户落地。