4. Debezium 安装与配置：Kafka Connect 环境搭建、Debezium MySQL Connector 配置、启动与验证

好，咱们进入正题。上一章聊了 CDC 的理论和选型，这一章我们直接动手。Debezium 的部署方式有好几种，但最主流、最抗造的，还是基于 Kafka Connect 的模式。我个人习惯在生产环境都用这套方案，原因很简单——解耦。你把数据变更流丢进 Kafka，下游想怎么消费都行，不会把源库压垮。

说白了，Debezium 就是一个 Kafka Connect 的 Source Connector。它假装自己是 MySQL 的从库，读 binlog，然后吐到 Kafka 里。咱们一步步来。

4.1 环境准备：你需要什么？

在开始之前，我建议你先确认一下手里的牌。我在项目中遇到过好几次，环境没对齐，折腾半天发现是 JDK 版本不对。

4.2 搭建 Kafka Connect 环境

Kafka Connect 是 Kafka 自带的一个分布式数据集成框架。你不需要额外安装什么，Kafka 包里就有。嗯，这里要注意，Connect 有两种运行模式：单机模式和分布式模式。咱们做生产，肯定用分布式。

4.2.1 下载与解压

去 Apache Kafka 官网下载二进制包，或者直接用 Confluent 的社区版。我个人倾向用 Confluent 的包，因为它把 Debezium 的插件都整合好了，省事。

# 下载 Kafka（以 3.2.0 为例）
wget https://archive.apache.org/dist/kafka/3.2.0/kafka_2.13-3.2.0.tgz
tar -xzf kafka_2.13-3.2.0.tgz
cd kafka_2.13-3.2.0

4.2.2 配置 Connect 分布式模式

配置文件在 config/connect-distributed.properties。我一般会改这几个关键参数：

# 集群标识，同一个集群的 Connect worker 必须一致
group.id=connect-cluster

# 存储偏移量、配置、状态的 topic 名称
offset.storage.topic=connect-offsets
config.storage.topic=connect-configs
status.storage.topic=connect-status

# 这些 topic 建议提前创建，分区数设 3~5，副本数按集群来
offset.storage.replication.factor=3
config.storage.replication.factor=3
status.storage.replication.factor=3

# 插件路径，后面放 Debezium 的 jar 包
plugin.path=/opt/kafka/plugins

4.2.3 启动 Kafka Connect

先确保 ZooKeeper 和 Kafka Broker 已经跑起来了。然后启动 Connect：

# 启动分布式模式
bin/connect-distributed.sh config/connect-distributed.properties

启动后，检查一下 REST API 是否正常：

curl http://localhost:8083/
# 返回 {"version":"3.2.0","commit":"..."} 就对了

4.3 安装 Debezium MySQL Connector

Debezium 的 connector 是以插件形式加载到 Kafka Connect 里的。说白了，就是把它的 jar 包扔到 plugin.path 目录下，然后重启 Connect。

4.3.1 下载 Connector 插件

你可以从 Debezium 官网下载，或者用 Confluent Hub 安装。我推荐后者，因为它会自动处理依赖。

# 用 Confluent Hub 安装（需要先安装 confluent-hub）
confluent-hub install debezium/debezium-connector-mysql:2.1.0

# 或者手动下载解压
wget https://repo1.maven.org/maven2/io/debezium/debezium-connector-mysql/2.1.0/debezium-connector-mysql-2.1.0-plugin.tar.gz
tar -xzf debezium-connector-mysql-2.1.0-plugin.tar.gz -C /opt/kafka/plugins/

解压后，/opt/kafka/plugins/ 目录下应该有一个 debezium-connector-mysql 文件夹，里面全是 jar 包。

4.3.2 验证插件是否加载成功

重启 Connect 后，调用 REST API 查看已安装的插件：

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

如果返回的列表里有 io.debezium.connector.mysql.MySqlConnector，说明加载成功了。如果没有，检查 plugin.path 配置和目录权限。我遇到过因为目录权限是 root，Connect 进程读不到的情况。

4.4 配置 MySQL 源库

在配置 Connector 之前，MySQL 那边得先准备好。Debezium 需要用一个有权限的账号来读 binlog。

4.4.1 创建 CDC 专用账号

我不建议用 root，风险太大。创建一个专用账号，只给必要的权限：

CREATE USER 'cdc_user'@'%' IDENTIFIED BY 'cdc_password';
GRANT SELECT, RELOAD, SHOW DATABASES, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'cdc_user'@'%';
FLUSH PRIVILEGES;

权限说明：

• SELECT：用于读取表结构做快照
• RELOAD：用于执行 FLUSH TABLES WITH READ LOCK，做一致性快照
• REPLICATION SLAVE, REPLICATION CLIENT：读 binlog 的必备权限

4.4.2 确认 binlog 配置

检查 MySQL 的配置文件 my.cnf：

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

4.5 配置并启动 Debezium MySQL Connector

好，环境都准备好了，现在来创建 Connector。通过 Kafka Connect 的 REST API 提交配置，这是最标准的方式。

4.5.1 创建 Connector 配置文件

创建一个 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": "cdc_password",
        "database.server.id": "223344",
        "database.server.name": "myapp-db",
        "database.include.list": "myapp",
        "table.include.list": "myapp.orders,myapp.users",
        "database.history.kafka.bootstrap.servers": "kafka1:9092,kafka2:9092",
        "database.history.kafka.topic": "schema-changes.myapp",
        "include.schema.changes": "true",
        "snapshot.mode": "initial",
        "tombstones.on.delete": "false"
    }
}

参数说明：

4.5.2 提交 Connector 配置

用 curl 提交给 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 http://localhost:8083/connectors/
# 返回 ["mysql-connector-01"]

4.6 验证数据是否正常捕获

Connector 启动后，Debezium 会先做一次快照，把当前表里的数据全量读到 Kafka。然后开始监听增量 binlog。怎么验证？

4.6.1 查看 Kafka Topic

Debezium 会为每个监听的表创建一个 topic，命名规则是 {database.server.name}.{database}.{table}。比如我们配置的 myapp-db.myapp.orders。

# 查看 topic 列表
bin/kafka-topics.sh --bootstrap-server localhost:9092 --list

# 消费 topic 看数据
bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 \
    --topic myapp-db.myapp.orders \
    --from-beginning

你会看到类似这样的 JSON 消息：

{
    "schema": {...},
    "payload": {
        "before": null,
        "after": {
            "id": 1001,
            "user_id": 42,
            "amount": 299.00,
            "status": "paid"
        },
        "source": {
            "version": "2.1.0",
            "connector": "mysql",
            "name": "myapp-db",
            "ts_ms": 1695000000000,
            "snapshot": "true",
            "db": "myapp",
            "table": "orders"
        },
        "op": "r",
        "ts_ms": 1695000000000
    }
}

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

4.6.2 模拟数据变更

在 MySQL 里执行一条 SQL：

INSERT INTO myapp.orders (user_id, amount, status) VALUES (43, 599.00, 'pending');

然后看 Kafka 消费者窗口，应该会立刻出现一条 op 为 c 的消息。如果看到了，恭喜你，CDC 链路通了。

4.7 常见问题与避坑

嗯，这里我总结几个我踩过的坑，你遇到了可以直接翻到这里查。

• Connector 状态显示 FAILED：去 Kafka Connect 的日志里看，多半是数据库连不上或者权限不够。检查 database.hostname 和密码。
• 快照超时：如果表很大，默认的快照超时时间可能不够。在配置里加 "snapshot.fetch.size": 10000 和 "snapshot.lock.timeout.ms": 60000。
• binlog 位置丢失：如果 Connector 重启后报错说找不到 binlog 文件，说明 MySQL 的 binlog 已经被清理了。调大 expire_logs_days，或者用 snapshot.mode: when_needed 让它自动重新快照。
• Schema 变更没捕获到：检查 include.schema.changes 是否设为 true，并且 database.history.kafka.topic 配置正确。

4.8 本章知识体系总览

最后，我用一张图把整个流程串起来。你对照着这张图，应该能清晰地看到数据从 MySQL 到 Kafka 的完整路径。

这张图把今天讲的内容都串起来了。从 MySQL 的 binlog，到 Debezium 解析，再到 Kafka 存储，最后给下游消费。你配置的时候，心里要时刻想着这条链路，哪个环节断了，就去查哪个环节的日志。