4、Debezium实战：MySQL Binlog配置、Debezium MySQL Connector配置、启动与监控、常见问题排查

好，咱们进入实战环节。前面讲了那么多理论，现在该动手了。这一章我会带着你，从MySQL的Binlog配置开始，一步步把Debezium跑起来。嗯，这里要注意，每一步都有坑，我会把我在项目中踩过的雷都告诉你。

4.1 MySQL Binlog 配置：这一步错了后面全白搭

说实话，我见过太多人卡在这一步。Binlog没开，或者格式不对，后面Debezium怎么折腾都连不上。咱们先把MySQL的Binlog配好。

4.1.1 检查当前Binlog状态

先登录MySQL，看看当前情况：

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

如果 log_bin 是 OFF，那就要改配置了。Binlog格式必须是 ROW，这是Debezium的硬性要求。为什么？因为只有ROW模式才能记录每一行数据的变化，Statement模式只记录SQL，对于CDC来说信息不够。

4.1.2 修改MySQL配置文件

找到MySQL的配置文件（通常是 /etc/my.cnf 或 /etc/mysql/mysql.conf.d/mysqld.cnf），在 [mysqld] 段落下添加：

[mysqld]
# 必须开启
log_bin = mysql-bin
# 必须为ROW
binlog_format = ROW
# 建议设置，保留7天
expire_logs_days = 7
# 建议设置，指定数据库（可选）
binlog_do_db = your_database_name
# 服务器ID，集群中必须唯一
server_id = 1

4.1.3 重启MySQL并验证

# 重启MySQL
systemctl restart mysql
# 或
service mysql restart

# 再次验证
SHOW VARIABLES LIKE 'log_bin';
SHOW VARIABLES LIKE 'binlog_format';

确认 log_bin 为 ON，binlog_format 为 ROW，这一步就过了。

4.2 创建CDC专用MySQL用户

我个人习惯给CDC单独创建一个用户，权限最小化，安全第一。别用root，那是给自己挖坑。

-- 创建用户
CREATE USER 'cdc_user'@'%' IDENTIFIED BY 'StrongPassword123!';

-- 授权：Debezium需要这些权限
GRANT SELECT, RELOAD, SHOW DATABASES, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'cdc_user'@'%';

-- 刷新权限
FLUSH PRIVILEGES;

解释一下这几个权限的用途：

权限	用途
SELECT	读取表结构、数据快照
RELOAD	执行 FLUSH TABLES WITH READ LOCK（快照时需要）
SHOW DATABASES	列出数据库
REPLICATION SLAVE	连接Binlog
REPLICATION CLIENT	获取Binlog位置信息

4.3 Debezium MySQL Connector 配置

好，MySQL那边准备好了。现在来配Debezium Connector。Debezium是跑在Kafka Connect上的，所以你得先有个Kafka Connect集群。假设你已经有了，咱们直接看Connector配置。

4.3.1 配置文件示例

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

{
  "name": "mysql-connector-01",
  "config": {
    "connector.class": "io.debezium.connector.mysql.MySqlConnector",
    "database.hostname": "192.168.1.100",
    "database.port": "3306",
    "database.user": "cdc_user",
    "database.password": "StrongPassword123!",
    "database.server.id": "1",
    "database.server.name": "my-app",
    "database.include.list": "your_database_name",
    "table.include.list": "your_database_name.users,your_database_name.orders",
    "database.history.kafka.bootstrap.servers": "kafka1:9092,kafka2:9092",
    "database.history.kafka.topic": "schema-changes.my-app",
    "include.schema.changes": "true",
    "snapshot.mode": "initial",
    "tombstones.on.delete": "false",
    "decimal.handling.mode": "double"
  }
}

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

• database.server.name：这是Kafka Topic的前缀。比如你设成 my-app，那表 users 的变更就会发到 my-app.your_database_name.users 这个Topic。
• snapshot.mode：initial 表示启动时先做一次全量快照，然后增量同步。如果你只要增量，可以设成 schema_only。
• tombstones.on.delete：设成 false 可以避免删除操作产生墓碑消息，减少Kafka存储压力。
• decimal.handling.mode：我习惯用 double，这样Decimal类型直接转成浮点数，下游处理方便。如果你要精确值，用 string。

4.3.2 提交Connector到Kafka Connect

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

返回 201 Created 就说明提交成功了。你可以用 GET /connectors/ 查看所有Connector，用 GET /connectors/mysql-connector-01/status 查看状态。

4.4 启动与监控：别等出事了再看日志

Connector提交后，Debezium会自动开始工作。但你不能扔那儿不管。我建议你从一开始就建立监控机制。

4.4.1 查看Connector状态

# 查看所有Connector
curl http://localhost:8083/connectors/

# 查看特定Connector状态
curl http://localhost:8083/connectors/mysql-connector-01/status

# 查看Connector配置
curl http://localhost:8083/connectors/mysql-connector-01/config

状态字段说明：

状态	含义
RUNNING	正常运行
PAUSED	已暂停
FAILED	出错了，赶紧看日志
UNASSIGNED	正在分配任务

4.4.2 查看Kafka Topic中的数据

用Kafka自带的命令行工具，看看数据有没有进来：

# 列出所有Topic
kafka-topics.sh --bootstrap-server localhost:9092 --list

# 消费Topic查看数据
kafka-console-consumer.sh --bootstrap-server localhost:9092 \
  --topic my-app.your_database_name.users \
  --from-beginning

如果你看到类似这样的JSON，说明数据已经同步了：

{
  "schema": {...},
  "payload": {
    "before": null,
    "after": {
      "id": 1,
      "name": "张三",
      "email": "zhangsan@example.com"
    },
    "source": {
      "version": "1.9.7",
      "connector": "mysql",
      "name": "my-app",
      "ts_ms": 1700000000000,
      "snapshot": "true",
      "db": "your_database_name",
      "table": "users"
    },
    "op": "r",
    "ts_ms": 1700000000000
  }
}

op 字段表示操作类型：r 是快照读取，c 是创建，u 是更新，d 是删除。

4.4.3 监控指标

Debezium暴露了JMX指标，我建议你用Prometheus + Grafana来监控。关键指标：

• StreamingQueueCurrentSize：当前队列大小，如果持续增长说明消费跟不上。
• MilliSecondsBehindSource：延迟毫秒数，这个最重要。
• TotalNumberOfEventsSeen：已处理事件总数。

4.5 常见问题排查：这些坑我都踩过

实战中肯定会遇到问题。我把最常见的几个列出来，你遇到了直接对照排查。

4.5.1 Connector启动失败：权限不足

现象：日志里报 Access denied for user 'cdc_user'@'...'。

原因：MySQL用户权限不够，或者密码插件不兼容。

解决：重新授权，确认用了 mysql_native_password。

4.5.2 快照阶段卡住不动

现象：Connector状态是RUNNING，但Kafka Topic里一条数据都没有。

原因：表数据量太大，或者 snapshot.fetch.size 设置太小。

解决：在Connector配置里加 "snapshot.fetch.size": "10000"，分批读取。如果表有几百G，考虑用 snapshot.mode: "schema_only" 跳过快照，只做增量。

4.5.3 Binlog位置丢失

现象：重启Connector后报 Binlog position ... is not available。

原因：MySQL的Binlog已经清理了，Debezium记录的偏移量找不到了。

解决：增大 expire_logs_days，或者用 snapshot.mode: "when_needed" 让Debezium自动重新做快照。

4.5.4 数据类型转换异常

现象：Kafka里的数据字段类型不对，比如时间戳变成了数字。

原因：Debezium对某些MySQL类型有默认的转换规则。

解决：在Connector配置里加 "converters": "datetime"，然后定义 datetime.type 参数。或者用 event.deserialization.failure.handling.mode: "warn" 先看看哪些字段有问题。

4.5.5 性能问题：延迟越来越高

现象：MilliSecondsBehindSource 持续增长。

原因：下游消费速度跟不上，或者Kafka Connect资源不够。

解决：

• 增加Kafka Connect的Worker节点。
• 调整 max.poll.records 和 fetch.max.bytes。
• 检查下游消费者是否有瓶颈。
• 考虑用 table.include.list 只监控必要的表。

4.6 小结

这一章的内容其实就三件事：配好MySQL、配好Connector、学会监控和排错。你只要把这几步走通了，Debezium就能稳定跑起来。嗯，说实话，第一次配置可能会遇到各种小问题，但别慌，对照上面的排查清单，一个一个过，总能解决。