3. Debezium入门

好，咱们正式开始聊 Debezium。说实话，CDC 工具不少，但 Debezium 是我个人用得最顺手的一个。它基于 Kafka，天生就是为流式数据准备的。你想想看，一个数据库的变更日志，能像消息队列一样被消费，这本身就是件很酷的事。

这一节，我会把 Debezium 的架构拆开来讲。包括它的三大组件——Connector、Snapshot、Streaming，还有两种部署模式——Embedded 和 Kafka Connect。最后，我们聊聊 Source Record 和 Offset 这两个核心概念。嗯，这些都是你上手 Debezium 必须搞明白的东西。

Debezium 整体架构

先看一张图，把 Debezium 的骨架搭起来。

这张图把 Debezium 的脉络画清楚了。从左到右：源数据库 → Debezium 核心 → Kafka。下面挂着部署模式和核心概念。咱们一个一个拆。

Connector：连接器的职责

Connector 是 Debezium 的「翻译官」。它负责跟数据库打交道，把数据库的 binlog、WAL 日志或者 oplog 翻译成 Debezium 能理解的事件。

每种数据库都有自己的 Connector。比如：

• MySQL Connector：读取 binlog，解析 row-based 事件。
• PostgreSQL Connector：读取 WAL（Write-Ahead Log），支持逻辑复制槽。
• MongoDB Connector：读取 oplog，或者 Change Streams。
• SQL Server Connector：读取 CDC 表或变更跟踪。

Connector 的配置通常长这样：

{
  "name": "mysql-connector-01",
  "config": {
    "connector.class": "io.debezium.connector.mysql.MySqlConnector",
    "database.hostname": "192.168.1.100",
    "database.port": "3306",
    "database.user": "debezium",
    "database.password": "debezium_pwd",
    "database.server.id": "184054",
    "database.server.name": "myapp-db",
    "database.include.list": "inventory",
    "table.include.list": "inventory.orders,inventory.customers",
    "snapshot.mode": "initial"
  }
}

这里有个关键参数 snapshot.mode，它决定了 Debezium 启动时怎么处理存量数据。咱们接着聊 Snapshot。

Snapshot：存量数据的「快照」

Snapshot 是 Debezium 的「初始化器」。当你第一次启动 Connector 时，它会对数据库做一次全量快照。说白了，就是把表里已有的数据全部读出来，转成 Change Event 发到 Kafka。

为什么要做 Snapshot？因为 CDC 只捕获增量变更。如果你不先做一次全量，那 Kafka 里只有新数据，历史数据就丢了。

Snapshot 有几种模式：

模式	说明	适用场景
initial	启动时执行全量快照，然后切换到 Streaming	默认模式，大多数场景用这个
when_needed	只在需要时执行快照（比如表结构变更后）	需要手动控制快照时机
never	跳过快照，直接进入 Streaming	只关心增量数据，或者已经通过其他方式同步过全量
schema_only	只捕获表结构，不读取数据	只需要 schema 变更事件

Snapshot 完成后，Debezium 会自动记录一个 Offset，标记「全量快照已完成」。然后无缝切换到 Streaming 模式。

Streaming：实时增量捕获

Streaming 是 Debezium 的「实时引擎」。它持续监听数据库的变更日志，把每一条 INSERT、UPDATE、DELETE 都转成 Change Event，实时推送到 Kafka。

Streaming 的核心机制是「轮询 + 解析」：

• MySQL：轮询 binlog 文件，解析 binlog event。
• PostgreSQL：通过逻辑复制槽接收 WAL 变更。
• MongoDB：监听 oplog 或 Change Streams。

Streaming 过程中，Debezium 会不断更新 Offset。这个 Offset 记录了当前读取到的日志位置。比如 MySQL 的 binlog 文件名 + 偏移量，PostgreSQL 的 LSN（Log Sequence Number）。

我曾经遇到过一个场景：生产环境的 MySQL 主库 binlog 保留时间只有 24 小时。Debezium 因为网络抖动断连了 30 小时，重启后 Offset 指向的 binlog 已经被清理了。结果就是——数据断层。所以，binlog 保留时间一定要大于你容忍的最大断连时间。我建议至少保留 72 小时。

部署模式：Embedded vs Kafka Connect

Debezium 支持两种部署方式。选哪个，取决于你的架构。

Embedded 模式

Embedded 模式是把 Debezium 作为一个 Java 库嵌入到你的应用里。你写代码启动 Connector，自己管理生命周期。

优点：

• 轻量，不需要额外部署 Kafka Connect 集群。
• 适合做「嵌入式 CDC」，比如在微服务里直接捕获变更。

缺点：

• 你需要自己处理 Offset 的持久化。
• 扩展性有限，单机部署。

代码示例：

// Embedded 模式启动 Debezium
DebeziumEngine<ChangeEvent<String, String>> engine = DebeziumEngine.create(Json.class)
    .using(props)
    .notifying(record -> {
        System.out.println("Key: " + record.key());
        System.out.println("Value: " + record.value());
    })
    .build();

// 启动引擎
ExecutorService executor = Executors.newSingleThreadExecutor();
executor.execute(engine);

Kafka Connect 模式

这是最主流的部署方式。Debezium 作为 Kafka Connect 的一个 Source Connector 运行。你只需要配置 JSON 文件，通过 REST API 提交给 Kafka Connect 集群。

优点：

• 分布式，高可用，支持水平扩展。
• Offset 由 Kafka Connect 自动管理，持久化到 Kafka 内部 topic。
• 有 REST API，方便运维。

缺点：

• 需要部署 Kafka Connect 集群，增加运维成本。
• 依赖 Kafka。

提交 Connector 的 REST API 调用：

curl -X POST http://localhost:8083/connectors \
  -H "Content-Type: application/json" \
  -d '{
    "name": "mysql-connector-01",
    "config": {
      "connector.class": "io.debezium.connector.mysql.MySqlConnector",
      "database.hostname": "192.168.1.100",
      "database.port": "3306",
      "database.user": "debezium",
      "database.password": "debezium_pwd",
      "database.server.id": "184054",
      "database.server.name": "myapp-db",
      "database.include.list": "inventory",
      "table.include.list": "inventory.orders",
      "snapshot.mode": "initial"
    }
  }'

核心概念：Source Record 和 Offset

这两个概念是 Debezium 的「灵魂」。搞懂了它们，你就掌握了 Debezium 的数据模型。

Source Record

Source Record 是 Debezium 输出的每一条变更事件。它包含三个部分：

• Key：主键或唯一键，用于标识被变更的行。
• Value：变更的详细内容，包括 before、after、source、op 等字段。
• Headers：元数据，比如数据库名、表名、操作类型。

一个典型的 Source Record（JSON 格式）长这样：

{
  "schema": { ... },
  "payload": {
    "before": {
      "id": 1001,
      "name": "Alice",
      "email": "alice@example.com"
    },
    "after": {
      "id": 1001,
      "name": "Alice",
      "email": "alice_new@example.com"
    },
    "source": {
      "version": "1.9.7.Final",
      "connector": "mysql",
      "name": "myapp-db",
      "ts_ms": 1695000000000,
      "snapshot": "false",
      "db": "inventory",
      "table": "customers",
      "server_id": 184054,
      "file": "mysql-bin.000003",
      "pos": 12345
    },
    "op": "u",
    "ts_ms": 1695000001000
  }
}

字段说明：

• op：操作类型。c=创建，u=更新，d=删除，r=快照读取。
• before：变更前的数据（UPDATE 和 DELETE 有）。
• after：变更后的数据（CREATE 和 UPDATE 有）。
• source：元数据，记录了这条变更来自哪个数据库、哪个表、binlog 的哪个位置。

Offset

Offset 是 Debezium 的「书签」。它记录了当前已经处理到数据库变更日志的哪个位置。下次重启时，Debezium 从 Offset 指向的位置继续读取，保证不丢不重。

Offset 的存储方式取决于部署模式：

部署模式	Offset 存储位置	说明
Kafka Connect	Kafka 内部 topic（connect-offsets）	自动管理，高可用
Embedded	文件系统（.offset 文件）或自定义存储	需要自己实现持久化

Offset 的内容示例（MySQL）：

{
  "file": "mysql-bin.000003",
  "pos": 12345,
  "row": 1,
  "server_id": 184054,
  "event": 2
}

这个 Offset 表示：「我已经读到 mysql-bin.000003 文件的第 12345 字节位置，第 1 行，第 2 个事件」。

小结

这一节我们把 Debezium 的骨架搭起来了。Connector 负责翻译，Snapshot 做全量初始化，Streaming 做实时增量。部署模式选 Kafka Connect 还是 Embedded，取决于你的架构复杂度。Source Record 和 Offset 是数据模型的核心，搞懂了它们，你就能理解 Debezium 的「断点续传」和「数据一致性」是怎么实现的。