4、Debezium实战：MySQL Binlog监听、Debezium Server配置、Kafka Connect集成

这一章我们直接动手。理论讲再多，不如跑通一条数据流来得实在。

Debezium 是什么？说白了，它就是数据库的「眼线」。它能盯着 MySQL 的 Binlog，一有变化就往外推。我最早接触它是在 2019 年，那时候公司要从 MySQL 实时同步数据到 Elasticsearch，试了几种方案都不太满意，直到遇到 Debezium——嗯，这玩意儿确实靠谱。

本章核心目标： 搭建一条从 MySQL → Debezium → Kafka 的实时数据管道。你会亲手配置 Binlog、启动 Debezium Server、再用 Kafka Connect 把它集成进生产链路。

4.1 MySQL Binlog 监听原理

先搞清楚一件事：MySQL 怎么知道数据变了？答案是 Binlog。

Binlog（Binary Log）记录了所有数据变更操作。INSERT、UPDATE、DELETE 都会写进去。Debezium 做的事情，就是伪装成一个 MySQL Slave，去读取 Binlog 的变更事件。

我个人习惯把 Binlog 的格式设为 ROW。为什么？因为只有 ROW 模式才能记录每一行数据变更前后的完整内容。STATEMENT 模式只记 SQL，MIXED 模式在某些场景下会丢字段细节——我在项目中踩过这个坑，查了半天才发现是 Binlog 格式的问题。

小提示： 生产环境务必使用 ROW 格式。同时开启 binlog_row_image=FULL，这样 Debezium 才能拿到完整的 before/after 数据。

配置 Binlog 的步骤很简单，在 MySQL 的 my.cnf 里加上这几行：

[mysqld]
server-id = 223344
log_bin = mysql-bin
binlog_format = ROW
binlog_row_image = FULL
expire_logs_days = 7

然后重启 MySQL。验证是否生效：

SHOW VARIABLES LIKE 'log_bin%';
SHOW VARIABLES LIKE 'binlog_format%';

看到 log_bin = ON 和 binlog_format = ROW，就说明配置成功了。

注意： MySQL 8.0 默认开启 Binlog，但很多云数据库默认是关闭的。如果你用的是 RDS 或类似服务，记得去控制台打开 Binlog 开关。

4.2 Debezium Server 配置与启动

Debezium 有两种部署方式：独立 Server 和 Kafka Connect 插件。我们先讲 Server 模式，因为它最简单，适合快速验证。

下载 Debezium Server 的压缩包，解压后你会看到 conf 目录。核心配置文件是 application.properties。我贴一份我常用的配置：

debezium.sink.type=kafka
debezium.sink.kafka.producer.bootstrap.servers=localhost:9092
debezium.sink.kafka.producer.key.serializer=org.apache.kafka.common.serialization.StringSerializer
debezium.sink.kafka.producer.value.serializer=org.apache.kafka.common.serialization.StringSerializer

debezium.source.connector.class=io.debezium.connector.mysql.MySqlConnector
debezium.source.offset.storage.file.filename=/data/debezium/offsets.dat
debezium.source.database.hostname=192.168.1.100
debezium.source.database.port=3306
debezium.source.database.user=debezium_user
debezium.source.database.password=debezium_pass
debezium.source.database.server.id=12345
debezium.source.database.server.name=my-mysql-instance
debezium.source.database.include.list=mydb
debezium.source.table.include.list=mydb.orders,mydb.users
debezium.source.snapshot.mode=initial

这里有几个关键点：

  • debezium.source.snapshot.mode：我建议第一次用 initial，它会先做一次全量快照，然后切换到增量监听。后续重启可以用 schema_only 或 when_needed。
  • offset.storage.file.filename：记录消费位置的偏移量文件。如果丢了，Debezium 会重新拉全量——我经历过一次，几百万数据重新同步，差点把 MySQL 打挂。
  • database.server.name：这个值会作为 Kafka Topic 的前缀。比如这里配了 my-mysql-instance，那么 Topic 名就是 my-mysql-instance.mydb.orders。

启动命令很简单：

./bin/run.sh

看日志，如果出现 Connected to MySQL 和 Snapshot completed，就说明跑起来了。

避坑指南： 我曾经在启动时遇到 Access denied for user，原因是 MySQL 用户权限不够。Debezium 需要 REPLICATION SLAVE、REPLICATION CLIENT 和 SELECT 权限。记得用这条 SQL 授权：GRANT REPLICATION SLAVE, REPLICATION CLIENT, SELECT ON *.* TO 'debezium_user'@'%';

4.3 Kafka Connect 集成

Server 模式适合测试，但生产环境我更推荐用 Kafka Connect。为什么？因为它有 REST API、支持动态扩缩容、还能和 Kafka 生态无缝集成。

Kafka Connect 分两种模式：Standalone 和 Distributed。我直接说结论——用 Distributed。Standalone 只适合单机测试，生产环境一旦挂了就是单点故障。

配置步骤分三步：

  1. 把 Debezium 插件放到 Kafka Connect 的插件目录
  2. 配置 connect-distributed.properties
  3. 通过 REST API 提交连接器

先看插件部署。下载 Debezium MySQL Connector 的插件包，解压后放到 Kafka 的 plugins 目录：

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

然后在 connect-distributed.properties 里指定插件路径：

plugin.path=/opt/kafka/plugins/debezium
bootstrap.servers=localhost:9092
group.id=debezium-connect-cluster
key.converter=org.apache.kafka.connect.json.JsonConverter
value.converter=org.apache.kafka.connect.json.JsonConverter
key.converter.schemas.enable=false
value.converter.schemas.enable=false
offset.storage.topic=connect-offsets
config.storage.topic=connect-configs
status.storage.topic=connect-status

启动 Kafka Connect：

./bin/connect-distributed.sh config/connect-distributed.properties

最后，通过 REST API 提交连接器配置。我个人习惯用 curl 或者 Postman：

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_user",
      "database.password": "debezium_pass",
      "database.server.id": "12346",
      "database.server.name": "mysql-orders",
      "database.include.list": "mydb",
      "table.include.list": "mydb.orders",
      "snapshot.mode": "initial",
      "topic.prefix": "mysql-orders"
    }
  }'

返回 201 Created 就成功了。你可以用 GET /connectors 查看所有连接器，用 GET /connectors/mysql-orders-connector/status 查看运行状态。

注意： Kafka Connect 的 REST API 默认监听 8083 端口。如果端口冲突，可以在配置里改 rest.port=8084。另外，生产环境建议开启认证，不然谁都能往你的集群里塞连接器。

4.4 数据格式与 Topic 结构

数据进入 Kafka 后，长什么样？我截一条实际的消息给你看：

{
  "schema": { ... },
  "payload": {
    "before": null,
    "after": {
      "id": 1024,
      "order_no": "ORD20241001",
      "amount": 299.00,
      "status": "PAID",
      "create_time": 1727712000000
    },
    "source": {
      "version": "2.5.0.Final",
      "connector": "mysql",
      "name": "mysql-orders",
      "ts_ms": 1727712001000,
      "snapshot": "false",
      "db": "mydb",
      "table": "orders",
      "server_id": 223344
    },
    "op": "c",
    "ts_ms": 1727712001002
  }
}

重点看 op 字段：

操作 op 值 说明
CREATE c 新增记录
UPDATE u 更新记录，before 和 after 都有值
DELETE d 删除记录，after 为 null
READ r 全量快照阶段的数据

Topic 的命名规则是 {topic.prefix}.{database}.{table}。比如上面配置的 mysql-orders.mydb.orders。每个表一个 Topic，结构清晰。

经验之谈： 我建议在消费端做一层 Schema Registry 的集成。虽然 JSON 格式方便调试，但生产环境用 Avro 或 Protobuf 更靠谱——字段类型不会丢，兼容性也好控制。

4.5 整体架构图

下面这张图展示了本章搭建的完整数据流：

Debezium 实时同步架构 MySQL Binlog (ROW 格式) Debezium Server / Kafka Connect Kafka Topic: mysql-orders.* Binlog 模拟 Slave CDC 事件 关键说明： • Debezium 伪装成 MySQL Slave，读取 Binlog 变更事件 • 每条变更记录包含 before/after/source/op 四个核心字段 • Kafka Topic 命名规则：{topic.prefix}.{database}.{table} 完整数据流 MySQL INSERT/UPDATE/DELETE → Binlog → Debezium 解析 → Kafka Topic → 下游消费

4.6 常见问题与排查思路

跑通这条链路不难，但生产环境总会遇到各种幺蛾子。我列几个我踩过的坑：

  • 连接器状态显示 FAILED：先看日志。最常见的原因是 MySQL 连接超时，或者 Binlog 位置被清理了。检查 expire_logs_days 配置，我建议至少保留 3 天。
  • 数据延迟很大：检查 Kafka 的生产者配置。acks=1 和 linger.ms=0 能降低延迟。如果还不行，看看 MySQL 的 Binlog 写入频率是不是太高了。
  • Topic 里出现大量重复数据：这通常是 offset 提交失败导致的。Kafka Connect 的 offset.storage.topic 如果分区数不够，或者副本数不足，就会丢 offset。我一般给这个 Topic 配 25 个分区、3 副本。
  • DDL 变更导致连接器挂掉：Debezium 对 DDL 的支持有限。如果表结构变了，比如新增字段，连接器可能会报错。我建议在变更表结构前，先暂停连接器，改完再重启。
调试技巧： 用 kafka-console-consumer 直接消费 Topic，能最快看到数据格式是否正确。命令：./bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic mysql-orders.mydb.orders --from-beginning

嗯，这一章的内容就到这里。你动手跑一遍，把 MySQL 的 Binlog 打开，启动 Debezium，再往表里插几条数据，看看 Kafka 里能不能收到消息。能收到，你就掌握了 CDC 实时同步最核心的一环。

← 上一章 📖 返回目录 下一章 →