3. Debezium深度解析

Debezium 这东西，说白了就是 CDC 领域的事实标准。我最早接触它是在 2019 年，那时候公司要从 Oracle 迁到 MySQL，业务数据不能停，还得实时同步到数仓。试了一圈工具，最后还是选了 Debezium。为什么？因为它基于 Kafka Connect，生态好，社区活跃，而且对数据库的侵入性极低。

这一章，我会把 Debezium 的核心拆开来讲。架构长什么样？Connector 怎么配？Snapshot 和 Streaming 两种模式到底怎么切换？嗯，咱们一个一个来。

3.1 Debezium 整体架构

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

Debezium 架构图 Debezium 核心架构 源数据库 MySQL / PostgreSQL / MongoDB / Oracle Debezium Connector 读取 binlog / WAL / oplog Kafka Connect 集群 分布式 / 单机模式 Apache Kafka Topic db.server.table 格式 下游消费者 Flink / 数仓 / 搜索引擎 Offset 管理 Kafka Connect offset topic 💡 关键点： 1. Connector 负责抓取日志，Kafka Connect 负责调度和容错 2. 每个表对应一个 Kafka Topic，Topic 名格式：serverName.schemaName.tableName

从图里能看出来，Debezium 不是单打独斗的。它寄生在 Kafka Connect 框架里，通过 Connector 去读数据库的变更日志。我个人习惯把这条链路分成四层：

  • 源数据库层：MySQL 的 binlog、PostgreSQL 的 WAL、MongoDB 的 oplog，这些都是 CDC 的原始数据来源。
  • Connector 层：每个数据库类型对应一个专用 Connector。它负责解析日志格式，把行变更转成统一的事件结构。
  • Kafka Connect 层：负责 Connector 的生命周期管理、任务分配、偏移量持久化。说白了，就是让 Connector 能跑在集群里，挂了能自动恢复。
  • Kafka 消息层：变更事件被序列化成 Avro/JSON/Protobuf，写入 Topic。下游想怎么消费都行。

💡 我的经验：刚开始用 Debezium 时，我总把 Connector 和 Kafka Connect 混为一谈。其实 Connector 是插件，Kafka Connect 是运行时。你配的是 Connector，但真正干活的是 Kafka Connect 进程。

3.2 Connector 配置详解

配置 Connector 是上手 Debezium 的第一步。我见过不少人在这一步栽跟头——配错了参数，要么连不上库，要么丢数据。咱们拿 MySQL Connector 举例，把关键参数过一遍。

3.2.1 核心配置项

参数名 说明 必填 默认值
connector.class Connector 全限定类名 是 —
database.hostname 数据库主机地址 是 —
database.port 数据库端口 是 3306
database.user 数据库用户名 是 —
database.password 数据库密码 是 —
database.server.name 逻辑名称，用于 Topic 前缀 是 —
database.include.list 要监听的数据库列表 否 全部
table.include.list 要监听的表列表 否 全部
snapshot.mode 快照模式 否 initial
offset.storage 偏移量存储后端 否 Kafka

下面是一个完整的 MySQL Connector 配置示例：

{
  "name": "mysql-connector-orders",
  "config": {
    "connector.class": "io.debezium.connector.mysql.MySqlConnector",
    "database.hostname": "192.168.1.100",
    "database.port": "3306",
    "database.user": "debezium_user",
    "database.password": "debezium_pass",
    "database.server.name": "prod_orders",
    "database.include.list": "orders_db",
    "table.include.list": "orders_db.orders, orders_db.order_items",
    "snapshot.mode": "initial",
    "snapshot.locking.mode": "minimal",
    "tombstones.on.delete": "false",
    "decimal.handling.mode": "double",
    "include.schema.changes": "true"
  }
}

⚠️ 避坑指南：我曾经在生产环境配漏了 database.server.name，结果 Topic 名全乱了。这个值会作为 Topic 前缀，一旦上线再改，历史数据全对不上。建议一开始就定好命名规范，比如 环境_业务线 的格式。

3.2.2 权限配置

Debezium 需要数据库用户有特定权限。拿 MySQL 来说：

  • SELECT：读取表结构，做快照
  • RELOAD：执行 FLUSH TABLES WITH READ LOCK，保证快照一致性
  • REPLICATION CLIENT：查看主从状态
  • REPLICATION SLAVE：读取 binlog

我习惯单独建一个用户，只给最小权限。别偷懒用 root，安全审计过不去。

3.3 Snapshot 模式：初次启动的基线数据

Connector 第一次启动时，数据库里已经有一堆数据了。Debezium 不会直接跳进 Streaming 模式——它会先做一次 Snapshot，把当前全量数据拍下来，然后再开始监听增量变更。

Snapshot 模式有几种，我列个表对比一下：

模式 行为 适用场景
initial 先做全量快照，再切到 Streaming 默认模式，首次启动
when_needed 只在需要时做快照（如偏移量丢失） 容错恢复
never 不做快照，直接开始 Streaming 已有基线数据，只接增量
schema_only 只拍表结构，不拍数据 只关心 DDL 变更
schema_only_recovery 恢复时只重建 schema 偏移量丢失但 schema 还在

Snapshot 的执行流程大致是这样的：

  1. Connector 获取全局读锁（如果配置了 minimal 模式，只锁相关表）
  2. 读取当前 binlog 位置
  3. 逐表扫描，把每一行数据作为 read 事件写入 Kafka
  4. 释放锁
  5. 记录偏移量，切换到 Streaming 模式

🔑 关键点：Snapshot 阶段产生的每条记录，op 字段是 r（read），而不是 c（create）。下游消费时可以通过这个字段区分是快照数据还是实时变更。

3.4 Streaming 模式：持续捕获增量变更

Snapshot 完成后，Connector 自动进入 Streaming 模式。这时候它不再扫描表，而是蹲在 binlog 上，等着新事件冒出来。

Streaming 模式的核心机制：

  • 轮询 binlog：每隔 poll.interval.ms（默认 500ms）拉一次新事件
  • 事件转换：把 binlog 事件转成 Debezium 的 Change Event 格式
  • 偏移量提交：每处理一批事件，就把当前 binlog 位置写到 offset topic
  • 容错：如果进程挂了，重启后从上次提交的偏移量继续读

一个典型的 Change Event 长这样：

{
  "schema": { ... },
  "payload": {
    "before": { "id": 1001, "status": "pending" },
    "after": { "id": 1001, "status": "shipped" },
    "source": {
      "version": "1.9.7",
      "connector": "mysql",
      "name": "prod_orders",
      "ts_ms": 1696000000000,
      "snapshot": false,
      "db": "orders_db",
      "table": "orders",
      "server_id": 123,
      "gtid": null,
      "file": "mysql-bin.000023",
      "pos": 45678,
      "row": 0
    },
    "op": "u",
    "ts_ms": 1696000001000,
    "transaction": null
  }
}

注意 op 字段：c 是创建，u 是更新，d 是删除。你想想看，下游只要解析这个字段，就能知道数据发生了什么变化。

💡 我的习惯：Streaming 模式下，我会把 tombstones.on.delete 设为 false。墓碑消息是 Kafka 的压缩机制用的，对 CDC 消费端没啥用，还占磁盘。关了它，Topic 能小不少。

3.5 Snapshot 与 Streaming 的衔接

这里有个容易踩的坑：Snapshot 和 Streaming 之间怎么保证数据不丢不重？

Debezium 的做法是：

  1. 在 Snapshot 开始前，先记录当前 binlog 位置（称为 binlog 锚点）
  2. Snapshot 过程中，所有新产生的变更都在锚点之后
  3. Snapshot 完成后，从锚点开始消费 Streaming 事件

这样一来，Snapshot 和 Streaming 的数据在时间线上是连续的。不会因为快照时间长而漏掉中间产生的变更。

⚠️ 我曾经踩过的坑：有一次 Snapshot 跑了 40 分钟，期间业务库有大量更新。我以为 Debezium 会自动合并，结果发现下游出现了重复数据。后来查文档才知道，Snapshot 和 Streaming 是独立的两条管道，如果下游没有做幂等处理，重复是必然的。解决方案是在消费端按 source.file + source.pos 做去重。

3.6 实战建议

最后，给几个我在生产环境总结出来的建议：

  • 先做 Schema Only 快照：如果表很大，先跑一次 schema_only 模式，确认 Connector 能正常启动，再切到 initial 做全量。
  • 监控偏移量延迟：用 Kafka Lag 指标监控 Connector 的消费进度。如果延迟持续增长，说明 Connector 处理能力跟不上写入速度。
  • 不要频繁重启 Connector：每次重启都会触发一次新的 Snapshot（如果配置了 initial 模式）。我见过有人每天重启一次，结果 Snapshot 永远跑不完。
  • 测试环境用 never 模式：开发测试时，如果已经有基线数据了，直接用 snapshot.mode=never，省去快照时间。
← 上一章 📖 返回目录 下一章 →