📚 课程目录第4章 Debezium入门

4. Debezium入门：架构、核心组件与快速启动

Debezium 这个名字，说白了就是「Debezium = Debezium」。嗯，开个玩笑。它基于 Kafka Connect 构建，专门用来捕获数据库的变更事件。我最早接触它是在 2019 年，当时要做一个用户行为实时分析平台，MySQL 的 binlog 解析搞得我头大。后来发现 Debezium 把这层脏活全包了，确实省心不少。

4.1 Debezium 的整体架构

Debezium 不是一个独立的服务，它是一组 Kafka Connect 的 Connector 插件。你想想看，Kafka Connect 负责管理 Connector 的生命周期，Debezium 负责把数据库的变更转成 Kafka 消息。两者配合，天衣无缝。

核心架构图：Debezium 在 Kafka Connect 中的位置

4.2 核心组件详解

Debezium 的核心组件其实就三个：Connector、Snapshot 和 Streaming。我习惯把它们比作「搬运工」、「搬家车」和「快递员」。

4.2.1 Connector

Connector 是 Debezium 的入口。你配置好连接信息、表名、偏移量存储等参数，它就开始干活。每个 Connector 对应一个数据库实例，但可以监听多个表。

我的经验： 配置 Connector 时，database.server.name 这个参数一定要取好。它决定了 Kafka Topic 的前缀。我见过有人用 my-db-1，结果 Topic 名变成 my-db-1.mydb.users，看着就别扭。建议用业务域名或项目名，比如 order-service。

4.2.2 Snapshot

Snapshot 是初始快照阶段。Connector 启动时，它会先读取数据库的当前全量数据，写入 Kafka。这个过程只执行一次，除非你手动触发重新快照。

为什么会需要 Snapshot？因为 binlog 只记录变更，不记录历史。如果你从头开始消费，没有全量数据做基础，后面的变更就拼不回去。说白了，Snapshot 就是给你一个「初始状态」。

注意： Snapshot 默认是阻塞式的。它会锁表（取决于数据库和隔离级别）。我曾经在生产环境启动 Connector，结果业务方反馈订单表被锁了 3 分钟。后来我改用 snapshot.mode: schema_only，只拉结构不拉数据，再配合业务低峰期做全量同步，才解决了问题。

4.2.3 Streaming

Streaming 是持续变更捕获阶段。Snapshot 完成后，Debezium 自动切换到 Streaming 模式，实时监听 binlog 或 WAL。每一条 INSERT、UPDATE、DELETE 都会被捕获并发送到 Kafka。

Streaming 的核心是「断点续传」。Debezium 会定期把消费到的 binlog 位置（offset）写入 Kafka 的 __consumer_offsets 或一个独立的 Topic。如果 Connector 重启，它会从上次记录的位置继续消费，不会丢数据，也不会重复消费太多。

组件	职责	触发时机	是否阻塞
Connector	管理连接、配置、生命周期	启动时	否
Snapshot	读取全量数据写入 Kafka	首次启动或手动触发	是（默认）
Streaming	实时捕获增量变更	Snapshot 完成后	否

4.3 支持的数据库

Debezium 官方支持的数据库有这些：

MySQL — 基于 binlog，支持 GTID
PostgreSQL — 基于 WAL，支持逻辑复制槽
MongoDB — 基于 oplog
Oracle — 基于 LogMiner 或 XStream
SQL Server — 基于 CDC 表
Db2 — 基于日志
Cassandra — 基于 CommitLog
Vitess — 基于 VStream

我个人用得最多的是 MySQL 和 PostgreSQL。MySQL 的 binlog 格式建议用 ROW 模式，不然 Debezium 解析不了。PostgreSQL 需要开启 wal_level = logical，并且创建发布（Publication）。

避坑指南： 我曾经在 MySQL 5.6 上跑 Debezium，结果发现 binlog 格式是 MIXED，部分 DDL 语句解析失败。后来统一改成 ROW 格式，问题解决。所以，MySQL 必须用 ROW 格式的 binlog，这是硬性要求。

4.4 快速启动一个 MySQL Connector

好，理论说完了，咱们动手。下面我会带你从零启动一个 MySQL Connector。假设你已经有了 Kafka 和 Kafka Connect 环境。

4.4.1 前提条件

Kafka 2.8+ 集群（单节点也行）
Kafka Connect 以分布式模式运行
MySQL 5.7+ 或 8.0，开启 binlog
MySQL 用户有 REPLICATION SLAVE、REPLICATION CLIENT、SELECT 权限

4.4.2 下载 Debezium Connector 插件

从 debezium.io 下载 debezium-connector-mysql 的 tar 包，解压到 Kafka Connect 的插件目录。比如：

mkdir -p /opt/kafka/plugins/debezium
tar -xzf debezium-connector-mysql-2.5.0.Final.tar.gz -C /opt/kafka/plugins/debezium

然后在 connect-distributed.properties 中配置：

plugin.path=/opt/kafka/plugins

重启 Kafka Connect 进程。

4.4.3 创建 Connector 配置

通过 Kafka Connect REST API 创建 Connector。我习惯用 curl 直接 POST JSON：

curl -X POST http://localhost:8083/connectors \
  -H "Content-Type: application/json" \
  -d '{
    "name": "mysql-orders-connector",
    "config": {
      "connector.class": "io.debezium.connector.mysql.MySqlConnector",
      "database.hostname": "192.168.1.100",
      "database.port": "3306",
      "database.user": "debezium",
      "database.password": "debezium_pass",
      "database.server.name": "order-service",
      "database.include.list": "order_db",
      "table.include.list": "order_db.orders,order_db.order_items",
      "database.history.kafka.bootstrap.servers": "localhost:9092",
      "database.history.kafka.topic": "schema-changes.order-service",
      "snapshot.mode": "schema_only",
      "include.schema.changes": "true"
    }
  }'

参数说明：

database.server.name — Kafka Topic 的前缀，建议用业务名
table.include.list — 只监听指定的表，格式：数据库名.表名
snapshot.mode — initial（全量+增量）、schema_only（只拉结构）、when_needed（按需）
database.history.kafka.topic — 存储 DDL 历史，用于恢复 Schema

4.4.4 验证 Connector 状态

创建成功后，检查状态：

curl http://localhost:8083/connectors/mysql-orders-connector/status

返回 "state": "RUNNING" 就说明成功了。然后去 Kafka 里看看 Topic：

kafka-topics.sh --bootstrap-server localhost:9092 --list | grep order-service

你会看到类似 order-service.order_db.orders 的 Topic。用 kafka-console-consumer 消费一下，就能看到 Debezium 输出的 JSON 消息了。

4.4.5 一条消息长什么样

Debezium 输出的消息包含 before 和 after 两个字段，分别代表变更前后的数据。比如一条 UPDATE：

{
  "schema": {...},
  "payload": {
    "before": {
      "id": 1001,
      "status": "PENDING",
      "amount": 299.00
    },
    "after": {
      "id": 1001,
      "status": "PAID",
      "amount": 299.00
    },
    "source": {
      "version": "2.5.0.Final",
      "connector": "mysql",
      "name": "order-service",
      "ts_ms": 1712345678000,
      "snapshot": "false",
      "db": "order_db",
      "table": "orders",
      "server_id": 1,
      "gtid": null,
      "file": "mysql-bin.000023",
      "pos": 45678,
      "row": 0
    },
    "op": "u",
    "ts_ms": 1712345678123
  }
}

op 字段表示操作类型：c 创建、u 更新、d 删除、r 快照读取。嗯，这里要注意，op: "r" 只在 Snapshot 阶段出现，Streaming 阶段只有 c/u/d。

重要提醒： Debezium 输出的消息默认是 JSON 格式，但如果你用 Avro 或 Protobuf 序列化，需要额外配置 Schema Registry。我个人建议生产环境用 Avro，因为 JSON 的 Schema 信息太冗余，每条消息都带 Schema 会浪费大量带宽。

4.5 小结

这一章我们聊了 Debezium 的架构、三个核心组件（Connector、Snapshot、Streaming）、支持的数据库，以及如何快速启动一个 MySQL Connector。你想想看，从数据库到 Kafka，中间只需要一个 JSON 配置，Debezium 就把 binlog 解析、断点续传、Schema 管理这些脏活全包了。