实时数据订阅与分发系统可以将业务数据源变更实时分发分发到消息总线上，并维护消息的统一格式，提供通用的客户端框架供消息生产者与下游业务接入。

一般能用于以下场景：

• 索引构建：MySQL到ES

• 缓存管理：MySQL到Redis或本地cache

• 数据库镜像

• 实时备份

• 价格变化等重要业务信息订阅

实时数据订阅与分发系统一般都有如下几个核心模块构成：

• Change Data Capture（变更数据抓取，CDC）：负责实时抓取业务数据源的变更消息；
• 消息中间件：支撑消息的分发与堆积；
• Client：为生产者与消费者提供统一的接入途径，解决序列化、offset管理、监控报警等共性问题。

本文以Databus（
https://github.com/linkedin/databus）为例，来说明一个实时数据订阅与分发系统的基本构成与原理。

Databus 是一个实时的低延迟数据抓取系统, 它抓取业务数据源的实时变更, 并发送到中继(Databus Relay), 下游业务从中继获得变更数据进行业务处理:

根据Linkdin的介绍, Databus有以下特性:

• 来源独立：Databus支持多种数据来源的变更抓取，包括Oracle和MySQL。

• 可扩展、高可用：Databus能扩展到支持数千消费者和事务数据来源，同时保持高度可用性。

• 事务按序提交：Databus能保持来源数据库中的事务完整性，并按照事务分组和来源的提交顺序交付变更事件。

• 低延迟：数据源变更完成后，Databus能在微秒级内将事务提交给消费者。

• 无限回溯：Databus对消费者支持无限回溯能力。当消费者需要产生数据的完整拷贝时（比如新的搜索索引), 直接进行一次全量回溯即可。

2. 系统设计

Databus的结构与工作流如下图:

• 通过CDC订阅数据库变更

• 将变更消息放入Relay的缓存队列

• 各个client对队列中的消息进行消费

我们可以看到，核心组件为五个部分:

1）DatabusEventProducer

负责实时数据抓取CDC, 针对MySQL数据源, 开源方案提供了基于OpenReplicator(一个Binlog解析框架)的方案。

2）SchemaRegistry

注册DatabusEvent对应的Schema, 所有DatabusEvent需要按Schema进行序列化, 并在消息中保持Schema信息。

3）DatabusRelay

基于Netty实现的一个Server, 内部维护高性能的缓存消息队列RingBuffer,作为订阅消息的内存消息中间件，保证了消息的有序性。

4）BootstrapService

BootStrapService是特殊的DatabusClient, 它将来自DatabusRelay中的所有数据写入MySQL, 当客户端需要无限回溯时, 便请求BootstrapService拉取历史数据。

有很多系统是将消息直接投递到kafka或者rocketMQ，就能同时实现了DatabusRelays和BootstrapService的功能。

5）ClientLib:

ClientLib就是消费客户端Client，用来实时接收变更消息。其中封装了一些数据抓取细节, 比如当回溯的SCN(System Change Number)在中继上不存在时自动请求BootstrapService, 回溯完成后切回中继。

3. 核心模块浅析

DatabusRelay

DatabusRelay模块可类比为基于内存实现的消息队列, 下面是DatabusRelay的结构图:

我们可以看到，DatabusRelay运行于Netty容器中。

同时，它会启动一系列EventProducer, 从数据源或其他Relays拉取实时增量数据并写入EventBuffers。

EventBuffers由多RingBuffer组成, RingBuffer通过mmap进行写盘持久化。这种设计下,使得EventProducer与DatabusRelay在同一个Netty容器中, 避免了rpc调用，效率更高。

所有的增量数据, 都有一个System Change Number(SCN), 这个SCN由EventProducer产生, 保证全局递增, DatabusRelay需要记录每个RingBuffer目前的MaxSCN（类似Kafka的offset）, 并使用MaxSCN Reader/Writer进行持久化。持久化方式是本地文件存储。

DatabusClient

DatabusClient用于消费来自DatabusRelay的数据, 它作为一个lib提供给需要接入的服务。下面是官方给出的DatabusClient架构图:

客户端代码以回调形式注册到DatabusClient上, 并声明自己关心的资源。

启动后, Client通过读取当前checkpoint, 假如checkpoint在Relay中不存在, 那么启动Relay Puller 和 Bootstrap Puller分别从Relay和Bootstrap Service拉取数据, 并写入本地EventBuffer, Dispatcher不断poll EventBuffer中的数据, 分发到Callback Driver上, 并通知Checkpoint Persistence Provider记录当前读取的checkpoint(即SCN)。

这样就能实现对订阅消息的全量回溯, 向客户端代码屏蔽Relay与Boostrap Service的差异。

4. 扩展性

在上面的DataBus Relay的架构图可以看到

Event Producer除了可以订阅数据源之外，还能订阅其他Relays，可以通过Relay Chaining进行扩展。在Follower Relay中使用RelayEventProducer, 从Master Relay拉取数据, 这两个Relay就组成了Master和Follower的链式结构。当然，这种设计会使得变更数据在多个Relay中冗余，有些浪费空间。

都看到最后了，原创不易，点个关注，点个赞吧～

知识碎片重新梳理，构建Java知识图谱：github.com/saigu/JavaK…（历史文章查阅非常方便）