RocketMQ 流数据库解析：如何实现一体化流处理？

作者：林清山（隆基）

前言：

从初代开源消息队列崛起，到 PC 互联网、移动互联网爆发式发展，再到如今 IoT、云计算、云原生引领了新的技术趋势，消息中间件的发展已经走过了 30 多个年头。

目前，消息中间件在国内许多行业的关键应用中扮演着至关重要的角色。随着数字化转型的深入，客户在使用消息技术的过程中往往同时涉及交叉场景，比如同时进行物联网消息、微服务消息的处理，同时进行应用集成、数据集成、实时分析等，企业需要为此维护多套消息系统，付出更多的资源成本和学习成本。

在这样的背景下，2022 年，RocketMQ 5.0 正式发布，相对于 RocketMQ 4.0，架构走向云原生化，并且覆盖了更多的业务场景。

背景

在《RocketMQ 流存储解析：面向流场景的关键特性与典型案例》一文中我们介绍了 RocketMQ 面向数据集成提供的流存储的能力，基于流存储和业界主流的分布式流计算引擎结合，比如 Flink、Spark，能为用户提供完整的流处理能力。然而，在某些场景下，我们有机会为用户提供更简化的流处理方案，不需要维护多套分布式系统，通过 RocketMQ 5.0，就能提供一体化的流处理。

本文第一部分，我们将从概念和宏观角度理解什么是流处理；第二部分，我们再回到 RocketMQ 5.0，介绍 RocketMQ 提供的轻量流处理引擎 RStreams，了解其特性和原理；第三部分，我们将介绍 RocketMQ 的流数据库 RSQLDB，如何通过流存储和流计算的深度结合，进一步降低流处理使用门槛。

流处理场景

让我们先来了解一下什么是流处理？流处理过程包括流数据摄入、流数据存储和流计算。

第一个概念是流数据，它是与批量数据、离线数据相对的。其特点在于数据源源不断的产生，并且有一定的顺序，从而形成一个无边界数据流，类似于现实世界中的河流。比如信用卡交易、股票交易、IoT 设备传感数据，都可以认为是流数据。

第二个概念是流存储，在《RocketMQ 流存储解析：面向流场景的关键特性与典型案例》一文中重点介绍过，这里简要回顾一下。流存储就是面向流式数据进行深度优化的存储系统，类似于日志（Log），提供按照分区、位点进行读写操作的能力，数据是持续追加且不可变的。典型的流存储有 RocketMQ、KAFKA、AWS 的 Kinesis Data Streams。

第三个概念是流计算，面向流式数据的计算引擎，它主要的特征是实时计算、低延迟，能够实现有状态计算，比较典型的流计算引擎有 Flink、Spark Streaming、Kafka 的 KStreams 等等。

那么，通常什么时候会用到流处理场景呢？相对于批处理——通常是天级别的计算延迟，流处理更侧重于需要实时响应的场景，比如信用卡欺诈检测、股票实时投资、工厂设备维护，还有舆情实时监控等等。

流计算技术分析

流处理过程主要包括流数据的摄入、存储以及流式计算三个环节。虽然摄入和存储也是重要环节，但本文将聚焦于流计算所需的技术能力。

流计算的数据流可以概括为三个步骤：数据输入，数据转换，数据输出。 我们通过一个简单的案例 WordCount 来解释流计算的技术要点。看右边这个图，数据输入是实时产生的语句流，我们希望能够按照时间窗口统计每个单词出现的次数，按时间输出统计结果。我们基于流计算引擎，只需要写少量的代码，比如左下角的图，即可完成任务。

从这个案例，我们可以总结出流计算引擎需要具备的关键能力：

首先，需要支持丰富的可重用算子，采用函数式编程的方式，提升流计算的开发效率。

其次，需要具备容错能力，在计算过程中节点宕机时，能够通过重启或者其他计算节点接管恢复计算。

再来，流数据往往是大规模的，比如 IoT 设备产生的大规模传感数据，往往超出单机的计算能力。流计算引擎要具备大规模并行计算能力。

最后，流计算的结果往往用于关键业务决策，流计算引擎要能做到在大规模并行、容错切换、资源调度等场景下，保障计算结果的正确性。

RStreams

1. RStreams 的特点

面向流处理场景，RocketMQ 5.0 提供了原生的轻量流计算引擎 RStreams，它有三个特点：

首先，只依赖 RocketMQ 的原生技术栈，基于 RocketMQ 的不同类型 Topic 实现数据流处理，适合轻量输出、边缘计算场景。

其次，它的用法也很轻量，不用搭建流计算平台，用户没有额外的运维负担，直接使用 RStreams 的 SDK 编写流计算逻辑，并内嵌到业务应用（或者微服务中）即可。

最后，它覆盖了主流场景的所有算子，具备完整的流计算能力。包括无状态算子，比如过滤、map 等等，以及有状态算子，如聚合计算、窗口计算等等。

2. RStreams - 数据流

对于一个流计算引擎来说，最关键的是要了解整个数据流的情况。虽然从使用角度看，流计算是一次输入、转换和一次输出，而实际的实现过程中，流计算是由多个更加原子的算子多次输入、计算、输出组合在一起，涉及复杂的数据流图。

RStreams 完全是基于 RocketMQ 的流存储能力来实现数据流，面向用户的输入、输出分别对应 Source Topic 和 Sink Topic，而中间件的计算过程要基于 State Topic（即 CompactTopic）来维护流计算的中间状态，在计算过程可能还需要进行数据交换，比如按照单词统计词频中会用到 KeyBy 算子，RStreams 是基于 Shuffle topic 来实现的。

数据交换 - Shuffle Topic

关于 Shuffle Topic，我们再简单看一下。还是以 WordCount 为例，我们希望每个句子切割成单词后，要按照单词统计频率，这就需要把同一个单词的数据放到同一个计算实例上计数。RStreams 的实现就是把单词作为 Key hash 到同一个队列，基于 RocketMQ 的消费负载算法就可以保障同一个单词都在一个计算实例上统计。这就是 RStreams 的数据交换机制。

3. RStreams - 状态管理

我们再来看 RStreams 的另一个关键技术点——状态管理。

状态管理有两种场景，一种是容错场景，这里只需要依赖 RocketMQ 队列位点重放能力实现 checkpoint 机制就可以恢复计算状态。

另一种场景是有状态计算的中间计算结果维护，RStreams 通过 RocksDB 作为本地状态管理器，提供高性能、低延迟的状态读写，同时也基于 RocketMQ 的 CompactTopic 维护远程状态，定期和本地状态同步。这样一来，当本地节点磁盘损坏或者计算节点重新调度后，还可以从一个统一的数据存储中心恢复状态，提高状态数据的可靠性。

有状态算子 - Windows 举例

我们以 WordCout 案例中的窗口计算为例，来了解 RStreams 的有状态算子状态维护。

这个案例里，首先通过 Shuffle Topic 完成单词的分组统计，单词词频统计是按照时间窗口刷新，所以这里状态维护用的 Key 是 Topic + Q + 窗口时间 + 单词，Value 是统计数量，定时刷新到 RStreams 的状态存储中。当出现宕机，进行容错恢复后，窗口中的数据不用从头重新计算，保障流计算的实时性。

4. RStreams - 大规模计算

RStreams 的大规模并行计算，直接复用 RocketMQ 的无限扩展能力和负载均衡机制。比如基于 RocketMQ 的数据分片，流存储可以实现无限扩展；基于 RocketMQ 的分片负载消费模式，流计算节点也可以实现无限扩展。

5. RStreams - 弹性伸缩

下面我们再详细了解一下 RStreams 弹性伸缩的过程。在《RocketMQ 流存储解析：面向流场景的关键特性与典型案例》一文中提到，对于 RStreams 计算调度主要依赖 RocketMQ 的消费者队列负载均衡机制，数据源的每个数据分片只会被一个 RStreams 的实例读取计算。发生扩缩容的时候，会按照负载均衡算法重新分配计算节点。

除此之外，涉及有状态计算的时候，RStreams 还需要依赖 Compact Topic 维护状态，Compact Topic 的队列分布需要和 SourceTopic 保持一致，这样一来，数据源和对应的状态存储就都会被同一个 RStreams 计算节点重新加载。比如下面这张图，在发生缩容的时候，SourceTopic 队列 2 的数据和状态都调度到 RStreams 实例 2，从 checkpoint 加载数据恢复计算。

RSQLDB

下面我们进入第三部分，RocketMQ 的流数据库形态 RSQLDB。先来了解一下什么是流数据库，流数据库是在流处理技术逐渐走向成熟、普惠过程中发展出来的。在流处理的初级阶段，流存储、流计算是分离的，使用方式主要通过 SDK API 编程来实现。现在到了流处理的普及阶段，流处理的门槛进一步降低，流表一体化理论也走向成熟，用户可以基于传统数据库的概念、声明式的 SQL 语句来完成业务的流处理，技术学习门槛降低，效率进一步提升。

那流数据库和传统数据库有什么区别呢？我们来看下面的表格。首先，传统数据库的操作对象是表格，是由外部动作发起的面向静态数据的一次性查询，在这个过程中数据是静态的；而流数据库是相反的，是源源不断的流数据触发持续查询，数据是动态的、主动的，查询是被动的。

下图就是一个典型的流数据库的使用方式，通过一条 SQL 就可以完成流的过滤、窗口计算、聚合计算等能力。

1. RSQLDB 概览

面向流处理发展趋势，RocketMQ 5.0 推出了 RSQLDB，这是一款基于标准 SQL 来进行持续查询动态表的流数据库，支持了大量传统数据库的使用模式，包括 DDL、DML、查询和各种函数。下图是 RSQLDB 的架构图，它也是基于 RocketMQ 的一体化技术，底层是 RocketMQ 的流存储 + RStreams 的流计算原子能力，在这些能力之上提供了 SQL 解析器，把用户 SQL 转化为物理的流处理过程，最上层提供了多种形态的客户端，包括 SDK、控制台和命令行工具。

2. RSQLDB 示例

最后我们来看一个 RSQLDB 的典型示例，这个案例想要把电影票购买信息流和购买人信息流合并，最后输出购买人信息 + 电影票信息的完整输出流。

首先，我们用 RSQLDB 创建两张表，把 topic 数据流转化为表的抽象。然后，创建一个视图，这个视图是来自购买人信息和电影票购买信息流的双流合并后的视图，基于购买人 ID 进行 Join。最后，我们再创建一个结果表，把这个双流 Join 的视图写入到这个输出到结果表里。

这样我们就通过一些简单的声明式的 SQL 语句来完成的一个双流 Join 的流处理过程。

总结

这篇文章，我们系统的介绍了流处理的场景，从数据的摄入、存储到计算，它的优势是提升数字化业务实时响应的能力。RocketMQ 原生的轻量流计算框架，是基于 RocketMQ 的整流存储，本文也带大家了解了实现一个完整的一体化处理的能力。最后，我们对于 RocketMQ 的流处理，做了进一步的升华，就是把 RocketMQ 的流存储跟流计算融为一体，提供流数据库的使用形态，降低流处理的使用门槛。

我们将持续为您带来深度剖析 RocketMQ 5.0 的系列文章，欢迎点击此处进入官网了解更多详情，也欢迎填写表单进行咨询：https://survey.aliyun.com/apps/zhiliao/bzT3AfPaq