4. Debezium实战：MySQL Binlog配置，Debezium MySQL Connector配置与启动，观察变更事件（Insert/Update/Delete）的JSON结构

好，咱们进入真正的实战环节。这一章，我会带你亲手把Debezium跑起来，从MySQL的Binlog配置，到Connector的启动，再到亲眼看到那些变更事件的JSON长什么样。说白了，就是让你知道数据到底是怎么从MySQL流出来的。

4.1 先搞定MySQL Binlog

Debezium要监听MySQL的变更，前提是MySQL得把Binlog打开。这就像你要录视频，得先把摄像头的电源插上。我刚开始接触CDC的时候，就吃过这个亏——配置了半天Connector，结果MySQL压根没开Binlog，日志里全是连接超时。嗯，从那以后，我第一件事就是检查这个。

4.1.1 检查当前状态

登录你的MySQL，先看看Binlog是不是开着的：

-- 查看是否开启Binlog
SHOW VARIABLES LIKE 'log_bin';
-- 查看Binlog格式
SHOW VARIABLES LIKE 'binlog_format';
-- 查看当前Binlog文件列表
SHOW BINARY LOGS;

如果 log_bin 是 OFF，那咱们就得动手配置了。我个人习惯把Binlog格式设为 ROW，因为只有行模式才能记录每一行数据的变化细节，Debezium也依赖这个。

4.1.2 配置MySQL的my.cnf

找到你的MySQL配置文件（通常在 /etc/my.cnf 或 /etc/mysql/my.cnf），在 [mysqld] 段落下加入这几行：

[mysqld]
# 开启Binlog
log_bin = mysql-bin
# 设置Binlog格式为ROW
binlog_format = ROW
# 设置服务器ID，集群中必须唯一
server_id = 1
# 保留Binlog的天数，建议根据磁盘空间调整
expire_logs_days = 7
# 记录所有数据库的变更，也可以指定具体库
binlog_do_db = your_database_name

4.1.3 创建Debezium专用用户

Debezium需要一个MySQL用户来读取Binlog和数据库元数据。别用root，太危险。我建议创建一个权限最小化的用户：

-- 创建用户
CREATE USER 'debezium'@'%' IDENTIFIED BY 'your_password';
-- 授予读取Binlog的权限
GRANT REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'debezium'@'%';
-- 授予读取数据库元数据的权限
GRANT SELECT ON your_database_name.* TO 'debezium'@'%';
-- 刷新权限
FLUSH PRIVILEGES;

这里有个坑：REPLICATION SLAVE 和 REPLICATION CLIENT 是全局权限，不能指定到某个库。而 SELECT 权限可以精确到库，这样Debezium只能读取你指定的数据库，安全多了。

4.2 启动Debezium MySQL Connector

Binlog准备好了，接下来就是配置和启动Connector。我假设你已经跑起了Kafka和Kafka Connect（如果还没搞定，可以参考前面的章节）。

4.2.1 下载Connector插件

从 Debezium官网 下载 debezium-connector-mysql 插件包，解压后放到Kafka Connect的插件目录下。我个人习惯放在 /opt/kafka/plugins 里，然后在 connect-distributed.properties 中配置：

plugin.path=/opt/kafka/plugins

重启Kafka Connect，让插件生效。你可以通过REST API检查插件是否加载成功：

curl -s http://localhost:8083/connector-plugins | jq .

如果看到 io.debezium.connector.mysql.MySqlConnector，那就对了。

4.2.2 配置Connector

创建一个JSON配置文件，比如 mysql-connector.json：

{
  "name": "mysql-connector-demo",
  "config": {
    "connector.class": "io.debezium.connector.mysql.MySqlConnector",
    "database.hostname": "localhost",
    "database.port": "3306",
    "database.user": "debezium",
    "database.password": "your_password",
    "database.server.id": "1",
    "database.server.name": "mysql-server-01",
    "database.include.list": "your_database_name",
    "table.include.list": "your_database_name.your_table_name",
    "database.history.kafka.bootstrap.servers": "localhost:9092",
    "database.history.kafka.topic": "schema-changes.mysql",
    "include.schema.changes": "true",
    "snapshot.mode": "initial"
  }
}

4.2.3 启动Connector

用REST API把配置提交给Kafka Connect：

curl -i -X POST -H "Accept:application/json" -H "Content-Type:application/json" \
  http://localhost:8083/connectors/ \
  -d @mysql-connector.json

如果返回 201 Created，恭喜你，Connector启动成功了。你可以用这个命令查看状态：

curl -s http://localhost:8083/connectors/mysql-connector-demo/status | jq .

状态应该是 RUNNING。如果报错，别慌，看看Kafka Connect的日志，大部分问题都是权限、网络或者配置项写错了。

4.3 观察变更事件的JSON结构

Connector跑起来了，现在咱们往MySQL里插几条数据，看看Debezium吐出来的JSON长什么样。我用Kafka自带的命令行消费者来消费topic：

kafka-console-consumer.sh --bootstrap-server localhost:9092 \
  --topic mysql-server-01.your_database_name.your_table_name \
  --from-beginning \
  --property print.key=true \
  --property print.value=true \
  --property key.deserializer=org.apache.kafka.common.serialization.StringDeserializer \
  --property value.deserializer=org.apache.kafka.common.serialization.StringDeserializer

4.3.1 Insert事件

在MySQL里执行一条插入：

INSERT INTO your_table_name (id, name, age) VALUES (1, '张三', 28);

你会看到类似这样的JSON（我简化了，实际会更长）：

{
  "schema": { ... },
  "payload": {
    "before": null,
    "after": {
      "id": 1,
      "name": "张三",
      "age": 28
    },
    "source": {
      "version": "1.9.7.Final",
      "connector": "mysql",
      "name": "mysql-server-01",
      "ts_ms": 1690000000000,
      "snapshot": "false",
      "db": "your_database_name",
      "table": "your_table_name",
      "server_id": 1,
      "file": "mysql-bin.000001",
      "pos": 12345
    },
    "op": "c",
    "ts_ms": 1690000000000,
    "transaction": null
  }
}

注意看 op 字段：c 代表Create（插入）。before 是 null，因为插入前没有旧数据。after 就是新插入的行。source 里记录了这条变更来自哪个Binlog文件和位置，方便你回溯。

4.3.2 Update事件

执行更新：

UPDATE your_table_name SET age = 29 WHERE id = 1;

输出：

{
  "payload": {
    "before": {
      "id": 1,
      "name": "张三",
      "age": 28
    },
    "after": {
      "id": 1,
      "name": "张三",
      "age": 29
    },
    "op": "u",
    ...
  }
}

op 变成了 u（Update）。before 是更新前的数据，after 是更新后的。这样你就能知道哪一列变了，从什么值变成什么值。我在做实时数仓的时候，经常用这个字段来做增量更新，只处理变化的部分。

4.3.3 Delete事件

执行删除：

DELETE FROM your_table_name WHERE id = 1;

输出：

{
  "payload": {
    "before": {
      "id": 1,
      "name": "张三",
      "age": 29
    },
    "after": null,
    "op": "d",
    ...
  }
}

op 是 d（Delete）。before 是被删除的数据，after 是 null。这里要注意：默认情况下，Debezium会把删除事件也发到Kafka，但如果你下游不需要删除数据，可以在Connector配置里加 tombstones.on.delete: false 来关闭。

4.4 核心流程一览

为了让你更直观地理解整个流程，我画了一张图：

4.5 避坑指南

最后，分享几个我踩过的坑，希望能帮你省点时间：

• Binlog文件太大撑爆磁盘：我曾经有个项目，Binlog一天涨了50G，忘了设 expire_logs_days，结果磁盘满了，MySQL直接挂了。建议根据数据量设置保留天数，并监控磁盘使用率。
• Connector启动后没数据：检查 database.include.list 和 table.include.list 的格式，必须是 数据库名.表名，大小写敏感。我遇到过因为表名大小写不一致，导致Connector找不到表。
• JSON里中文乱码：确保MySQL的字符集是 utf8mb4，Debezium默认用UTF-8编码，如果源库是latin1，中文就会变成乱码。建表时记得加 CHARACTER SET utf8mb4。
• Kafka Topic分区数不够：如果数据量很大，单个分区会成为瓶颈。建议在Connector配置里设置 tasks.max 大于1，并提前规划好Topic的分区数。