CDC变更数据捕获：Canal与Debezium

可视化演示

展示CDC（变更数据捕获）核心机制：数据流管道模式可视化MySQL binlog经Canal/Debezium解析后通过Kafka投递至下游消费者（ES/缓存/仓库），架构对比模式展示Canal与Debezium的六层架构差异与关键指标对比。

CDC原理概述

你经营着一家大型物流仓库，每天有成千上万个包裹入库、改址、退货。客户最关心的不是仓库有多大，而是“我的包裹现在到哪了”。如果每个人都打电话来问，客服早就崩溃了。好在仓库有一本“扫描台账”——每一件包裹被扫码枪扫过，都会按顺序记上一行。真正聪明的做法是安排一个调度员坐在台账旁边：只要台账出现新的一行，就立刻抄送给关心这个包裹的人。这个“盯着台账、照抄变化、实时广播”的机制，就是 CDC（Change Data Capture，变更数据捕获）。

CDC 是一种从数据库日志中捕获数据变更、并把变更推送给下游系统的技术。它不修改业务代码，不要求业务方在写库的同时“顺便”通知别人，而是让数据库自己留下的日志成为唯一事实来源。对 AI 从业者而言，这相当于给数据平台接上了一根“神经”：实时特征、模型样本、缓存更新、搜索索引，全都依赖这根神经传递的变化信号。

在日志解析方案流行之前，业界常见两种笨办法。一是全表轮询：每隔几秒 SELECT 一次全表，和上次对比找出变化，白白消耗数据库 IO，延迟还高。二是应用双写：业务代码里同时写数据库和消息队列，侵入性强、容易漏写。相比之下，日志解析方案直接读取 MySQL binlog——数据库本身就把每次变更写进了日志，我们只是去“照抄”，因而低侵入、低延迟、高可靠，还能拿到变更前后的数据镜像。

就像调度员抄写台账的速度决定了客户收到通知的快慢，CDC 方案的好坏，第一个要看的就是“延迟”——从数据库提交事务，到下游真正拿到消息，中间隔了多久。延迟越低，业务越接近“实时”；延迟一旦抖动，往往意味着解析、网络或消费端出现了瓶颈。我们把这份延迟写成一个公式，它贯穿整条链路的设计。

binlog日志解析

为了让每一笔包裹动态都有据可查，仓库规定：所有扫码记录必须按时间顺序写进台账，不许撕页、不许插队。MySQL 也有一本这样的“台账”——binlog（二进制日志），它按顺序记录所有改变数据的操作，包括插入、更新、删除，甚至建表。它不是给人看的，而是给机器“追剧”用的：日积月累，就是数据库的一生。

你可能早就听说过 MySQL 主从复制：主库写 binlog，从库把 binlog 拉到本地“重放”，于是从库和主库拥有相同的数据副本。有趣的是，Canal、Debezium 正是搭了这趟顺风车：它们把自己伪装成 MySQL 的一个从库，向主库申请 binlog 的读取权限，但拿到日志后并不重放 SQL，而是把日志解析成一条条结构化的“变更事件”，转卖给下游。

binlog 有三种记录格式：statement 记录 SQL 语句本身，日志小但结果不可控；row 模式记录每一行数据变更的“前镜像”和“后镜像”，精确无损，是 CDC 工具最爱的格式；mixed 则是两者混合。在主从复制里，“从库读到哪一行”由位点（position）标记；在 CDC 里，同样靠位点实现断点续传——下次启动，从上次读到的位置继续。

台账不是无限期的，binlog 也默认只保留若干天（例如 expire_logs_days）。这意味着 CDC 擅长的是“从今天开始往后的每一笔”，而不是“把十年前的历史也找回来”。历史数据要用全量快照或备份来补齐——这个分工，后面的同步策略会细说。

Canal架构设计

现在给仓库正式配备一个调度组，名字叫 Canal。这个组内部有明确分工：有人守在台账前逐行抄录（Parser），有人把抄录内容翻译成统一格式的通知单（Sink），有人把通知单暂时存放在中转盒中（Store），最后还有专门的联络员把通知单送给外面的客户（Client）。四个人各司其职，整条抄送流水线就不会堵。

Canal 是阿里巴巴开源的中间件，核心机制就是伪装成 MySQL 从库，向主库发送 dump 协议请求，持续读取 binlog。它的基本运行单元叫 instance，内部由三大模块组成：EventParser 负责连接主库、拉取并解析 binlog；EventSink 负责过滤、加工和分发事件；EventStore 负责在内存中暂存解析好的事件，等待客户端来取。

光有流水线还不够，还要保证“消息不丢”。Canal 采用 ack/rollback 机制：客户端消费完一条消息后回传 ack，Canal 才推进位点；如果消费失败，可以 rollback 让消息留在 Store 里重新投递。另外，Canal 支持集群高可用：通过 ZooKeeper 协调多台 server，同一个 instance 同一时刻只有一台在解析日志，故障时自动切换，客户几乎无感知。

一句话总结，Canal 是一支为 MySQL 量身定制的“抄送班组”，性能强劲、与 Java 生态配合默契，但和 MySQL 绑定较紧。如果你的数据源不止 MySQL，或者想要一套更“标准化”的通知协议，那就该认识另一支队伍——Debezium。

Debezium架构

假设你的公司同时租用了好几家仓库，还对接了无数外部系统。这时你会希望所有通知沿用统一格式，不管包裹来自哪个仓库，客户收到的通知单长一个样。Debezium 就是这支讲究“标准协议”的调度队。它把自己建立在 Kafka Connect 之上，把所有数据库的变化都编码成一种通用记录格式，谁拿到都能直接消费。

Debezium 是一个开源分布式平台，通过“连接器（Connector）”支持 MySQL、PostgreSQL、MongoDB、Oracle 等多种数据库。每个连接器负责盯一个数据源，以 Kafka Connect Source Connector 的身份运行，把捕获到的变更持续写入 Kafka 的 topic。对使用者来说，部署变成了“启动一个连接器”，而不是维护一套自研工具。

Debezium 输出的记录称为 Source Record，核心字段包括：source（数据库、表名、binlog 位点等来源信息）、op（操作类型，c 表示新增、u 表示更新、d 表示删除、r 表示快照记录）、ts_ms（事件发生的时间戳）以及 before/after（变更前后的数据）。它还内置快照机制：首次启动会自动做一次全量快照，随后无缝切换成增量监听，整个过程对业务透明。

对比 Canal 和 Debezium：前者轻量、贴近 MySQL、性能调优空间大；后者更标准化、多数据库支持好、与 Kafka 生态天然融合。选型没有绝对优劣，关键看你的数据源是“纯 MySQL 电商后端”还是“多源异构的数据湖或微服务”。无论选谁，都绕不开一个问题：历史数据怎么补？增量从哪开始？这就是下一节的同步策略。

数据同步策略

接一个新客户系统，老板要求：把所有包裹当前状态同步过去，并且此后一直保持最新。聪明的做法分两步：先安排一次彻底盘点，把此刻仓库里所有包裹登记造册；然后让调度员持续跟踪新的进出记录。麻烦的是，盘点过程里包裹还在不断进出——怎么保证盘点结束时不重不漏？这时需要给两条流水线约定一个“交接点”。

第一步全量同步，通常用快照工具把当前数据整体搬走，同时记录快照那一刻 binlog 的位点；第二步增量同步，从那个位点开始持续消费 binlog 中新发生的变更并写入目标。这个“交接点”就是位点（position / offset），如同台账上的书签，精确记录“我已经读到哪一行”。全量与增量以位点为界，无缝衔接。

位点管理直接决定同步的可靠程度。Canal 以客户端 ack 为前提推进位点；Debezium 则把 offset 定时提交到 Kafka 内部 topic，重启后从已提交位点继续。由于网络和重启都可能带来重复投递，业界普遍采用“至少一次投递 + 下游幂等”的组合：下游按主键做 upsert 或去重，同一事件处理两遍结果一样，从而做到不丢不重。

大白话就是：盘点负责“把存量搬过去”，调度员负责“把增量追上来”，位点是两者之间唯一的接力棒。接力棒保存得好，系统重启一百次也不怕；保存得含糊，轻则重复消费，重则静默丢数据。下一节，我们把这些零件装进一条完整的实时链路，看看它怎么在真实世界转起来。

实时链路实践

至此，整个通知中心已经成型：扫码枪产生记录 → 调度员抄写翻译 → 放进中转站 → 客户取走通知单。对应到技术栈，就是 MySQL 写 binlog → Canal / Debezium 解析 → 消息队列（Kafka / RocketMQ）缓冲分发 → 下游应用消费。每个环节职责单一，任何一环故障都不会拖垮整条链路。

消息队列是这条链路的中转站。Canal 可以通过内置 adapter 把事件写入 Kafka 或 RocketMQ，Debezium 则天生以 Kafka Connect Source 的身份写入 topic。下游按主题订阅，像订杂志一样各取所需：订单组只订订单 topic，用户组只订用户 topic。消费端组成消费组水平扩展，处理完一条提交一次 offset，提交成功才代表这条消息“消化”了。

故障恢复是链路设计的试金石。下游服务挂了不要紧，消息留在队列里，offset 停在那里，等恢复后继续消费即可；同步进程整个宕机，重启后从持久化位点继续读日志，最多重复、不会丢失。真正需要盯紧的是监控指标：lag（消费积压量）、解析吞吐、错误率、位点是否停滞。lag 一旦持续上涨，说明消费能力跟不上生产速度，就得扩容或优化。

回头看这条链路：数据库每一次细微变动，都被日志忠实记录，被调度工具实时捕获，被消息队列稳稳传递，最终被下游系统及时消化——从“客户打电话问包裹”到“包裹一动全公司都知道”，这就是 CDC 给分布式系统带来的秩序感。而这份秩序，正是数据平台工程师在 AI 时代最值钱的基本功。

AI 应用场景

Python 演示

"""Simulate a CDC pipeline: read binlog events, parse, push to MQ."""import json  class BinlogReader:    """Simulate MySQL binlog; yield change events one by one."""     def __init__(self, events):        self.events = events        self._pos = 0     def read_event(self):        if self._pos >= len(self.events):            return None        event = self.events[self._pos]        self._pos += 1        return event  class CanalParser:    """Simulate Canal; parse raw events and filter by table."""     def __init__(self, tables=None):        self.tables = list(tables or [])        self.gc_count = 0     def parse(self, event):        if self.tables and event["table"] not in self.tables:            return None        self.gc_count += 1        return {            "schema": event["db"],            "table": event["table"],            "eventType": event["type"],            "data": event["data"],            "old": event.get("old"),        }  class MessageQueue:    """Simulate a MQ buffer between producer and consumer."""     def __init__(self):        self.items = []     def publish(self, msg):        self.items.append(msg)     def consume(self):        if not self.items:            return None        return self.items.pop(0)  def main():    events = [        {"db": "shop", "table": "orders", "type": "INSERT",         "data": {"id": 1, "amount": 99, "status": "paid"}},        {"db": "shop", "table": "orders", "type": "UPDATE",         "data": {"id": 1, "status": "shipped"}, "old": {"status": "paid"}},        {"db": "shop", "table": "orders", "type": "DELETE",         "data": {"id": 2, "amount": 50, "status": "cancel"}},    ]    reader = BinlogReader(events)    parser = CanalParser(tables=["orders"])    mq = MessageQueue()     while True:        raw = reader.read_event()        if raw is None:            break        msg = parser.parse(raw)        if msg is not None:            mq.publish(json.dumps(msg))            print("pushed:", msg["eventType"], msg["table"], msg["data"])     print("queue size:", len(mq.items))    while True:        msg = mq.consume()        if msg is None:            break        print("consumed:", msg)  if __name__ == "__main__":    main()

1. class BinlogReader:模拟 MySQL 的 binlog 数据源，就像仓库的扫描台账。
2. self._pos = 0记录日志读到哪一行，等价于 binlog 的位点（position）。
3. def read_event(self):每次调用返回一条变更记录，模拟 Canal 按顺序读取日志。
4. class CanalParser:模拟 Canal 的解析模块，负责把原始日志变成结构化消息。
5. self.tables = list(tables or [])保存允许转发的表白名单，对应 Canal 的过滤配置。
6. def parse(self, event):按表名过滤并转换事件，等价于 Canal 的 filter + parser。
7. if self.tables and event["table"] not in self.tables:只转发关注表的变更，避免下游收到无关数据。
8. mq.publish(json.dumps(msg))把结构化消息序列化，等价于投递给 MQ 的字节流。
9. def consume(self):模拟下游消费者从队列拉取变更。
10. if __name__ == "__main__":可执行入口，直接运行即可看到推送与消费的完整日志。

常见误区

面试题