环境准备：从零搭建实时同步基座

说实话，很多同学学 Debezium 卡住，不是因为概念难懂，而是环境没搭对。我见过太多人花了两天时间配置 MySQL Binlog，结果漏了一个参数，数据就是抓不到。这一章，咱们把地基打牢。

我会带着你一步步搞定四个核心组件：MySQL Binlog 配置、Zookeeper 部署、Kafka 集群搭建、Debezium Connector 安装。每个环节我都会告诉你「为什么这么做」，以及「我踩过的坑」。

2.1 MySQL Binlog 配置：开启数据捕获的开关

Binlog 是 MySQL 的二进制日志，记录所有数据变更。Debezium 就是靠读它来捕获数据的。我刚开始做 CDC 时，以为只要在 my.cnf 里加一行 log-bin=mysql-bin 就行，结果发现少配了两个参数，折腾了一下午。

下面是我验证过的配置，直接复制到你的 MySQL 配置文件（通常是 /etc/my.cnf 或 /etc/mysql/mysql.conf.d/mysqld.cnf）：

[mysqld]
# 必须：开启 Binlog
log-bin=mysql-bin
# 必须：Binlog 格式设为 ROW
binlog-format=ROW
# 必须：指定 MySQL 实例的唯一 ID
server-id=1
# 强烈建议：保留 Binlog 的天数
expire_logs_days=7
# 可选：Binlog 文件大小上限
max_binlog_size=1G
# 可选：记录所有数据库（可指定具体库）
binlog-do-db=your_database_name

配置完成后，重启 MySQL：

sudo systemctl restart mysql

验证 Binlog 是否开启：

mysql> SHOW VARIABLES LIKE 'log_bin';
+---------------+-------+
| Variable_name | Value |
+---------------+-------+
| log_bin       | ON    |
+---------------+-------+

mysql> SHOW BINARY LOGS;
+------------------+-----------+
| Log_name         | File_size |
+------------------+-----------+
| mysql-bin.000001 |       154 |
+------------------+-----------+

看到 ON 和日志列表，说明 Binlog 已经跑起来了。

2.2 Zookeeper 部署：Kafka 的「大管家」

Zookeeper 是 Kafka 的协调服务。说白了，它负责管理 Kafka 集群的元数据、Leader 选举、故障转移。没有它，Kafka 集群就转不起来。

我个人习惯用单节点 Zookeeper 做开发和测试，生产环境至少 3 节点。这里先讲单节点部署。

步骤 1：下载并解压

wget https://archive.apache.org/dist/zookeeper/zookeeper-3.8.1/apache-zookeeper-3.8.1-bin.tar.gz
tar -xzf apache-zookeeper-3.8.1-bin.tar.gz
cd apache-zookeeper-3.8.1-bin

步骤 2：配置 zoo.cfg

cp conf/zoo_sample.cfg conf/zoo.cfg

编辑 conf/zoo.cfg，核心参数如下：

步骤 3：启动 Zookeeper

bin/zkServer.sh start

检查状态：

bin/zkServer.sh status
# 输出：Mode: standalone

2.3 Kafka 集群搭建：消息的「高速公路」

Kafka 负责接收 Debezium 捕获的变更事件，并持久化到 Topic 中。下游消费者（比如 Flink、Spark、你的业务服务）再从 Topic 里拉取数据。

步骤 1：下载并解压

wget https://downloads.apache.org/kafka/3.4.0/kafka_2.13-3.4.0.tgz
tar -xzf kafka_2.13-3.4.0.tgz
cd kafka_2.13-3.4.0

步骤 2：配置 server.properties

编辑 config/server.properties，重点修改以下参数：

# 每个 Broker 的唯一 ID
broker.id=0
# 监听地址（0.0.0.0 表示所有网卡）
listeners=PLAINTEXT://0.0.0.0:9092
# Zookeeper 连接地址
zookeeper.connect=localhost:2181
# 日志保留时间（小时）
log.retention.hours=168
# Topic 副本数（单节点设为 1）
default.replication.factor=1

步骤 3：启动 Kafka

bin/kafka-server-start.sh config/server.properties &

验证 Kafka 是否正常运行：

# 创建一个测试 Topic
bin/kafka-topics.sh --create --topic test-topic --bootstrap-server localhost:9092 --partitions 1 --replication-factor 1

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

看到 test-topic 出现在列表里，说明 Kafka 已经就绪。

2.4 Debezium Connector 安装：CDC 的「翻译官」

Debezium Connector 是连接 MySQL 和 Kafka 的桥梁。它读取 Binlog，解析成结构化的变更事件，然后写入 Kafka Topic。

步骤 1：下载 Debezium MySQL Connector

wget https://repo1.maven.org/maven2/io/debezium/debezium-connector-mysql/2.3.0.Final/debezium-connector-mysql-2.3.0.Final-plugin.tar.gz
tar -xzf debezium-connector-mysql-2.3.0.Final-plugin.tar.gz

步骤 2：配置 Kafka Connect

Kafka Connect 是 Kafka 自带的连接器运行框架。编辑 config/connect-distributed.properties：

# 插件路径（指向 Debezium 解压目录）
plugin.path=/path/to/debezium-connector-mysql
# Kafka 集群地址
bootstrap.servers=localhost:9092
# 内部 Topic 配置
group.id=connect-cluster
key.converter=org.apache.kafka.connect.json.JsonConverter
value.converter=org.apache.kafka.connect.json.JsonConverter
key.converter.schemas.enable=true
value.converter.schemas.enable=true

步骤 3：启动 Kafka Connect

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

验证 Connector 是否加载成功：

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

返回结果里应该包含 io.debezium.connector.mysql.MySqlConnector。

步骤 4：注册 Debezium Connector

创建一个 JSON 配置文件 register-mysql.json：

{
  "name": "mysql-connector",
  "config": {
    "connector.class": "io.debezium.connector.mysql.MySqlConnector",
    "database.hostname": "localhost",
    "database.port": "3306",
    "database.user": "debezium",
    "database.password": "debezium",
    "database.server.id": "1",
    "database.server.name": "my-mysql",
    "database.include.list": "your_database_name",
    "database.history.kafka.bootstrap.servers": "localhost:9092",
    "database.history.kafka.topic": "schema-changes.my-mysql"
  }
}

注册 Connector：

curl -X POST -H "Content-Type: application/json" --data @register-mysql.json http://localhost:8083/connectors

检查 Connector 状态：

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

看到 "state": "RUNNING"，恭喜你，CDC 链路已经通了。

bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic my-mysql.your_database_name.your_table_name --from-beginning

2.5 环境验证清单

为了确保万无一失，我整理了一份检查清单。每完成一项，打个勾：

环境准备好了，接下来就可以真正开始使用 Debezium 了。记住，基础打牢了，后面才走得稳。