📚 课程目录第4章 Debezium实战

4. Debezium实战：架构解析、连接器配置与MySQL到Kafka的实时数据流搭建

好，咱们进入实战环节。Debezium 这个工具，说白了就是 CDC 领域的一把瑞士军刀。我最早接触它是在一个金融风控项目里，需要实时捕获核心交易库的变更，然后推送给下游做反欺诈分析。当时选型对比了好几个方案，最后还是 Debezium 胜出——原因很简单：它对 MySQL binlog 的解析做得最成熟，社区也最活跃。

这一章，我会带着你从架构入手，然后一步步搭出一条从 MySQL 到 Kafka 的实时数据管道。你跟着走一遍，基本就能应付大部分生产场景了。

4.1 Debezium 架构：三个核心组件

Debezium 的架构其实不复杂。它基于 Kafka Connect 构建，核心就三个角色：连接器（Connector）、转换器（Converter） 和 Kafka Connect 运行时。

一句话总结：Debezium 连接器负责从数据库拉取变更事件，Kafka Connect 负责把事件写入 Kafka，转换器负责把事件序列化成你需要的格式。

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

4.2 连接器配置：一步步来

配置 Debezium MySQL 连接器，核心就是写一个 JSON 配置文件。我习惯用 Kafka Connect 的 REST API 来提交，这样不用重启集群。

4.2.1 基础配置项

先看一个最简配置：

{
  "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_pass",
    "database.server.id": "223344",
    "database.server.name": "dbserver1",
    "database.include.list": "mydb",
    "table.include.list": "mydb.orders,mydb.users",
    "database.history.kafka.bootstrap.servers": "kafka01:9092,kafka02:9092",
    "database.history.kafka.topic": "schema-changes.mydb",
    "include.schema.changes": "true",
    "snapshot.mode": "initial"
  }
}

这里有几个关键参数，我重点说一下：

database.server.id：每个连接器必须唯一。Debezium 会模拟一个 MySQL slave 去拉 binlog，这个 ID 就是 slave 的 server_id。我曾经见过两个连接器配了同一个 ID，结果 MySQL 主库直接报错，binlog dump 线程冲突。
database.server.name：这是 Kafka Topic 的前缀。比如配成 dbserver1，那 orders 表的变更就会写入 dbserver1.mydb.orders。
snapshot.mode：首次启动时要不要做全量快照。我一般用 initial，它会先做一次快照，然后无缝切换到 binlog 流。

我的习惯：生产环境里，我会把 snapshot.mode 设成 when_needed，配合 snapshot.locking.mode 为 minimal，这样能减少对源库的锁影响。但如果你对数据一致性要求极高，还是用 initial 更稳妥。

4.2.2 高级配置：处理 DDL 和表结构变更

Debezium 通过 database.history 机制来追踪表结构变化。它会将 DDL 语句写入一个专门的 Kafka Topic。这样即使连接器重启，也能从 Topic 里恢复表结构信息。

{
  "config": {
    "database.history.kafka.bootstrap.servers": "kafka01:9092",
    "database.history.kafka.topic": "schema-changes.mydb",
    "include.schema.changes": "true",
    "schema.history.internal": "io.debezium.storage.kafka.history.KafkaSchemaHistory"
  }
}

嗯，这里要注意：schema.history.internal 这个参数在 Debezium 2.0 之后改过名字，老版本叫 database.history。如果你用的是新版本，记得用新参数名，否则配置不生效。

4.3 搭建实时数据流：从 MySQL 到 Kafka

好，理论说完了，咱们动手搭一条完整的管道。我会用 Docker Compose 来演示，这样你本地也能跑起来。

4.3.1 环境准备

你需要准备三个服务：MySQL、Kafka（含 Zookeeper）、Kafka Connect（含 Debezium 插件）。

这是我的 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
    depends_on:
      - zookeeper
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1

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

  connect:
    image: debezium/connect:2.4
    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"

注意：MySQL 必须开启 binlog，且格式为 ROW。如果你用的是云数据库（比如 RDS），记得在参数组里把 binlog_format 改成 ROW，binlog_row_image 改成 FULL。我遇到过有人用 MINIMAL 模式，结果捕获到的数据只有主键，其他字段全是 null——排查了半天才发现是这个问题。

4.3.2 启动连接器

服务启动后，用 REST API 提交连接器配置：

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

提交成功后，你可以检查连接器状态：

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

返回 RUNNING 就说明一切正常。

4.3.3 验证数据流

现在，在 MySQL 里插入一条数据：

INSERT INTO mydb.orders (id, customer, product, amount) 
VALUES (1, '张三', '笔记本电脑', 5999.00);

然后消费 Kafka Topic 看看：

docker exec -it kafka kafka-console-consumer \
  --bootstrap-server kafka:9092 \
  --topic dbserver1.mydb.orders \
  --from-beginning

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

{
  "schema": { ... },
  "payload": {
    "before": null,
    "after": {
      "id": 1,
      "customer": "张三",
      "product": "笔记本电脑",
      "amount": 5999.00
    },
    "source": {
      "version": "2.4.0.Final",
      "connector": "mysql",
      "name": "dbserver1",
      "ts_ms": 1700000000000,
      "snapshot": "false",
      "db": "mydb",
      "table": "orders",
      "server_id": 1,
      "gtid": null,
      "file": "mysql-bin.000001",
      "pos": 1234
    },
    "op": "c",
    "ts_ms": 1700000001000
  }
}

看到 "op": "c" 了吗？这代表这是一个 Create 事件。其他操作类型：u 是 Update，d 是 Delete，r 是快照读取（Snapshot）。

关键字段说明：

payload.before：变更前的数据（Delete 事件有值，Create 事件为 null）
payload.after：变更后的数据（Create/Update 事件有值）
payload.source：元数据，包含 binlog 文件名、位置、时间戳等
payload.op：操作类型，c/u/d/r

4.4 避坑指南：我踩过的几个坑

做 CDC 管道，有些坑你迟早会遇到。我直接列出来，你提前避开：

问题	现象	原因	解决方案
连接器启动失败	REST API 返回 500	MySQL 用户权限不足	给用户授权 `REPLICATION SLAVE, REPLICATION CLIENT`
数据延迟大	Kafka 消息比 MySQL 变更晚几秒	binlog 消费线程数不够	调大 `database.binlog.thread.count`（默认 1）
表结构变更后连接器挂掉	日志报 Schema 不匹配	DDL 事件未正确写入 history topic	检查 `database.history.kafka.topic` 是否存在，且 cleanup.policy 不能是 compact
快照阶段锁表	业务写入被阻塞	默认 snapshot 会加全局读锁	设置 `snapshot.locking.mode` 为 `minimal`

我曾经在一个生产环境里，因为没给 MySQL 用户授权 REPLICATION CLIENT，导致连接器启动后一直报 Access denied。排查了半小时才发现是权限问题——嗯，这种低级错误，犯过一次就记住了。

4.5 小结

这一章我们走通了 Debezium 的完整链路：从架构理解，到连接器配置，再到实际搭建 MySQL → Kafka 的实时管道。你掌握了这些，就已经能应对大部分 CDC 接入场景了。

说白了，Debezium 的核心价值就是「把数据库的变更，变成 Kafka 的消息」。剩下的工作——比如数据清洗、格式转换、路由分发——都可以在 Kafka Streams 或下游消费者里完成。