📚 课程目录第3章 Debezium入门

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

Debezium 这个名字，我第一次接触是在一个金融项目的实时风控场景里。当时业务方要求「数据库一有变更，下游系统必须在秒级感知」。我调研了一圈，最后选了 Debezium。说实话，它把 CDC 的复杂度封装得相当好，你不需要自己写 Binlog 解析逻辑，也不用操心断线重连这些破事。

这一章，我带你把 Debezium 的骨架摸清楚。从架构设计到核心组件，再到怎么把它跑起来。

3.1 Debezium 整体架构

Debezium 本质上是一组 Kafka Connect 的 Source Connector。它寄生在 Kafka Connect 框架里，负责从数据库的日志里捞变更事件，然后吐到 Kafka 主题上。

我画了一张架构图，你看一眼就明白了：

整个链路很清晰：数据库产生日志 → Debezium Connector 读取并解析 → 通过 Kafka Connect 写入 Kafka → 下游消费者拉取。我习惯把 Debezium 看作一个「日志翻译官」，它把数据库底层的二进制日志，翻译成结构化的 JSON 事件。

3.2 核心组件拆解

Debezium 的核心组件就三个：Connector、Snapshot 和 Streaming。咱们一个一个说。

3.2.1 Connector（连接器）

Connector 是 Debezium 的「大脑」。每个 Connector 对应一个数据库实例或一张表。你通过 Kafka Connect 的 REST API 提交一个 Connector 配置，它就开始干活了。

配置里要指定三样东西：

连接信息：数据库地址、端口、用户名、密码
监听范围：监听哪些数据库、哪些表
偏移量管理：记录当前读到日志的哪个位置

举个例子，一个 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",
    "database.password": "debezium_pwd",
    "database.server.id": "184054",
    "database.server.name": "dbserver1",
    "database.include.list": "order_db",
    "table.include.list": "order_db.orders,order_db.order_items",
    "database.history.kafka.bootstrap.servers": "kafka1:9092,kafka2:9092",
    "database.history.kafka.topic": "schema-changes.order_db"
  }
}

我的经验：database.server.name 这个参数很重要。它决定了 Kafka 主题的前缀。比如你设成 dbserver1，那订单表的变更事件就会发到 dbserver1.order_db.orders 这个主题上。命名规范一点，下游消费的时候会省很多事。

3.2.2 Snapshot（快照）

Snapshot 是 Debezium 启动时的「初始化阶段」。你想想看，一个 Connector 刚部署上去，它不知道数据库里已经有哪些数据。所以它需要先做一次全量快照，把当前所有存量数据读出来，发到 Kafka 里。

Snapshot 有几种模式：

模式	行为	适用场景
initial	先做全量快照，再切换为增量流	默认模式，首次部署
when_needed	只在需要时触发快照	恢复或重新捕获
never	跳过快照，直接开始增量流	已有历史数据，只关心增量
initial_only	只做快照，完成后停止	一次性数据迁移

我记得有一次在生产环境部署，表里有 2 亿行数据。Snapshot 跑了将近 40 分钟，把 Kafka 集群的带宽都打满了。后来我学乖了，大表做快照之前，先评估一下数据量和网络带宽，必要时用 snapshot.fetch.size 参数控制每次拉取的行数。

3.2.3 Streaming（增量流）

Snapshot 做完之后，Debezium 自动进入 Streaming 模式。它会持续监听数据库的事务日志，每来一条变更，就生成一个事件发到 Kafka。

Streaming 的核心机制是「偏移量追踪」。Debezium 会记录它读到了日志的哪个位置（比如 MySQL 的 Binlog 文件名 + 偏移量），定期提交到 Kafka 的一个特殊主题里。如果 Connector 重启了，它会从上次记录的位置继续读，不会丢数据，也不会重复读。

  关键点：Debezium 保证「至少一次」语义。极端情况下（比如 Kafka 集群宕机），可能会有少量重复事件。下游消费端要做好幂等处理。

3.3 支持的数据库类型

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

数据库	Connector 类名	依赖的日志机制	我踩过的坑
MySQL	MySqlConnector	Binlog (ROW 格式)	必须开启 Binlog，且格式设为 ROW
PostgreSQL	PostgresConnector	WAL (逻辑复制槽)	复制槽不清理会导致 WAL 膨胀
MongoDB	MongoDbConnector	Oplog / Change Streams	副本集必须开启 Oplog
Oracle	OracleConnector	LogMiner / XStream	需要额外授权，配置较复杂
SQL Server	SqlServerConnector	CDC 表 + 变更跟踪	需要先启用数据库级别的 CDC
Db2	Db2Connector	IBM 的 CDC 表	用得少，文档相对少

注意：MySQL 的 Binlog 格式必须设为 ROW。我曾经接手过一个项目，DBA 把 Binlog 格式设成了 STATEMENT，结果 Debezium 只能捕获到「影响了多少行」，拿不到具体变更的数据。折腾了半天才发现是格式问题。

3.4 环境搭建与配置

搭建 Debezium 环境，说白了就是搭一套 Kafka + Kafka Connect 集群，然后把 Debezium 的插件放进去。我一般用 Docker Compose 做快速验证，生产环境用 Kubernetes 或手动部署。

3.4.1 快速启动（Docker Compose）

这是我个人习惯的快速启动方式。一个 docker-compose.yml 搞定所有依赖：

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
    ports:
      - "9092:9092"
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1

  mysql:
    image: debezium/example-mysql:2.5
    ports:
      - "3306:3306"
    environment:
      MYSQL_ROOT_PASSWORD: debezium

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

启动命令就一行：

docker-compose up -d

等所有容器启动后，检查 Kafka Connect 是否就绪：

curl http://localhost:8083/connectors

返回一个空数组 []，说明 Connect 跑起来了。

3.4.2 注册 Connector

接下来注册一个 MySQL Connector。用 POST 请求把配置发过去：

curl -i -X POST http://localhost:8083/connectors \
  -H "Content-Type: application/json" \
  -d '{
    "name": "inventory-connector",
    "config": {
      "connector.class": "io.debezium.connector.mysql.MySqlConnector",
      "database.hostname": "mysql",
      "database.port": "3306",
      "database.user": "root",
      "database.password": "debezium",
      "database.server.id": "184054",
      "database.server.name": "dbserver1",
      "database.include.list": "inventory",
      "database.history.kafka.bootstrap.servers": "kafka:9092",
      "database.history.kafka.topic": "schema-changes.inventory"
    }
  }'

注册成功后，你可以查看 Connector 的状态：

curl http://localhost:8083/connectors/inventory-connector/status

如果返回 "state": "RUNNING"，恭喜你，Debezium 已经开始工作了。

3.4.3 验证数据同步

连上 MySQL，往 inventory 库里插一条数据：

docker exec -it mysql bash -c 'mysql -u root -pdebezium inventory'
mysql> INSERT INTO customers (first_name, last_name, email) VALUES ('张三', '李四', 'zhangsan@example.com');

然后消费 Kafka 主题看看：

docker exec -it kafka bash -c 'kafka-console-consumer \
  --bootstrap-server kafka:9092 \
  --topic dbserver1.inventory.customers \
  --from-beginning'

你会看到类似这样的 JSON 事件：

{
  "schema": {...},
  "payload": {
    "before": null,
    "after": {
      "id": 1001,
      "first_name": "张三",
      "last_name": "李四",
      "email": "zhangsan@example.com"
    },
    "source": {
      "db": "inventory",
      "table": "customers"
    },
    "op": "c",
    "ts_ms": 1700000000000
  }
}

"op": "c" 表示这是一条 Create 操作。还有 u（更新）、d（删除）、r（快照读取）。

一个小技巧：如果你只想看增量数据，不想看 Snapshot 阶段的历史数据，可以在 Connector 配置里加 "snapshot.mode": "never"。这样 Debezium 启动后直接进入 Streaming 模式，只捕获启动之后的变更。

3.5 避坑指南

最后分享几个我实际项目中遇到的坑：

Binlog 保留时间：MySQL 的 Binlog 默认只保留几天。如果 Debezium 宕机时间超过这个窗口，重启后会找不到 Binlog 位置，只能重新做 Snapshot。建议把 expire_logs_days 设大一点，或者用 binlog.expire.logs.seconds 控制。
PostgreSQL 复制槽泄漏：如果 Connector 被删除但复制槽没清理，WAL 文件会一直堆积，撑爆磁盘。我一般会在删除 Connector 前手动查一下复制槽并清理。
Kafka 主题分区数：Debezium 默认每个表一个主题，分区数用 Kafka 的默认值（1 个）。如果表的数据量很大，建议提前创建主题并设置足够的分区数，否则消费端会成为瓶颈。

嗯，这一章的内容就到这。Debezium 的入门其实不复杂，核心就是理解 Connector、Snapshot、Streaming 这三个组件怎么配合。你把环境搭起来，跑一遍增删改查，感受一下事件的结构，后面学高级特性就顺了。