3. Debezium入门：架构、核心组件与环境搭建

说实话，Debezium 是我在 CDC 领域用得最顺手的工具之一。它基于 Kafka Connect 构建，天生就是为数据同步而生的。我记得第一次在生产环境部署 Debezium 时，心里还有点打鼓——毕竟它要直接读取数据库的 binlog，万一搞崩了怎么办？后来用熟了才发现，只要配置得当，它比想象中要稳得多。

这一节，咱们就从头捋一捋 Debezium 的核心逻辑。你不需要一次性记住所有细节，先搞懂它怎么工作、有哪些零件、怎么搭起来，后面实战才不会懵。

3.1 Debezium 的整体架构

Debezium 本质上是一组 Kafka Connect 的 Source Connector。它把数据库的变更事件转换成 Kafka 消息，然后下游的消费者（比如 Flink、Spark、或者另一个数据库）就能拿到这些变更。

它的架构其实很简单，我画了一张图帮你理解：

你看，数据从数据库出来，经过 Debezium Connector 变成 Kafka 消息，再通过 Kafka Connect 写入 Topic。下游想怎么消费都行。这个链路虽然看起来长，但每个环节都是解耦的，出了问题也好排查。

3.2 核心组件详解

3.2.1 Connector —— 连接器的灵魂

Connector 是 Debezium 的入口。每个数据库类型对应一个 Connector，比如 io.debezium.connector.mysql.MySqlConnector。它的职责很简单：告诉 Kafka Connect 该监听哪个数据库、哪个表、怎么处理事件。

我个人习惯把 Connector 的配置分成三块：

• 数据库连接信息：host、port、user、password
• 监听范围：database.include.list、table.include.list
• 行为控制：snapshot.mode、tombstones.on.delete

这里有个坑，我曾经遇到过：如果数据库用户权限不够，Connector 连 binlog 都读不到。所以记得给用户加上 REPLICATION SLAVE 和 REPLICATION CLIENT 权限。

3.2.2 Snapshot —— 初始快照

当你第一次启动 Connector 时，它不会直接开始监听变更。它会先做一次全量快照（Snapshot），把当前数据库里的数据全部读出来，写入 Kafka。

为什么要这么做？你想想看，如果只监听增量，那之前的数据就丢了。Snapshot 就是为了解决这个「历史数据」的问题。

Debezium 支持几种 Snapshot 模式：

模式	行为	适用场景
initial	启动时做一次全量快照，然后切到增量	默认模式，大多数场景
when_needed	只在需要时做快照（比如 schema 变更后）	已有存量数据，但需要重新同步
never	不做快照，直接开始监听增量	只关心新数据，不关心历史
schema_only	只同步表结构，不同步数据	只想拿到 schema，数据后续再补

3.2.3 Streaming —— 实时流式捕获

Snapshot 做完之后，Connector 会自动切换到 Streaming 模式。它会持续监听数据库的变更日志（比如 MySQL 的 binlog），每来一条变更，就生成一条 Kafka 消息。

Streaming 的核心是 offset。Debezium 会记录当前读到 binlog 的哪个位置，即使 Connector 重启，也能从断点处继续读，不会丢数据。

我记得有一次，生产环境的 Kafka 集群挂了半小时，恢复后 Debezium 自动从断点续传，一条数据都没丢。嗯，这就是 offset 机制的功劳。

3.3 环境搭建实战

光说不练假把式。咱们现在就把环境搭起来。你需要准备：

• 一个 MySQL 实例（5.7 或 8.0）
• 一个 Kafka 集群（单节点就够了）
• 一个 Kafka Connect 服务（distributed 模式）

下面是我常用的 Docker Compose 配置，你直接复制就能用：

version: '3.8'
services:
  zookeeper:
    image: confluentinc/cp-zookeeper:7.4.0
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181

  kafka:
    image: confluentinc/cp-kafka:7.4.0
    depends_on:
      - zookeeper
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092

  mysql:
    image: mysql:8.0
    environment:
      MYSQL_ROOT_PASSWORD: rootpass
      MYSQL_DATABASE: testdb
    command: --server-id=1 --log-bin=mysql-bin --binlog-format=ROW

  connect:
    image: debezium/connect:2.5
    depends_on:
      - kafka
      - mysql
    environment:
      BOOTSTRAP_SERVERS: kafka:9092
      GROUP_ID: 1
      CONFIG_STORAGE_TOPIC: connect-configs
      OFFSET_STORAGE_TOPIC: connect-offsets
      STATUS_STORAGE_TOPIC: connect-status
    ports:
      - "8083:8083"

启动之后，用 curl 注册一个 MySQL Connector：

curl -i -X POST -H "Accept:application/json" \
  -H "Content-Type:application/json" \
  http://localhost:8083/connectors/ \
  -d '{
    "name": "mysql-connector",
    "config": {
      "connector.class": "io.debezium.connector.mysql.MySqlConnector",
      "database.hostname": "mysql",
      "database.port": "3306",
      "database.user": "root",
      "database.password": "rootpass",
      "database.server.id": "1",
      "database.server.name": "dbserver1",
      "database.include.list": "testdb",
      "table.include.list": "testdb.users",
      "snapshot.mode": "initial",
      "offset.storage": "org.apache.kafka.connect.storage.FileOffsetBackingStore",
      "offset.storage.file.filename": "/tmp/offsets.dat",
      "offset.flush.interval.ms": "60000"
    }
  }'

3.4 验证是否跑通

注册成功后，你可以用 Kafka 自带的命令行工具看看 Topic 里有没有数据：

# 查看 Topic 列表
kafka-topics --bootstrap-server localhost:9092 --list

# 消费消息
kafka-console-consumer --bootstrap-server localhost:9092 \
  --topic dbserver1.testdb.users --from-beginning

如果看到类似这样的 JSON 输出，说明 Debezium 已经成功捕获到数据了：

{
  "schema": {...},
  "payload": {
    "before": null,
    "after": {
      "id": 1,
      "name": "张三",
      "email": "zhangsan@example.com"
    },
    "source": {
      "db": "testdb",
      "table": "users"
    },
    "op": "c",
    "ts_ms": 1700000000000
  }
}

这里的 op 字段表示操作类型：c 是创建，u 是更新，d 是删除。你往 MySQL 里插一条数据，这边马上就能看到。