📚 课程目录第2章环境准备

环境准备：Kafka 集群搭建、ZooKeeper 配置、MySQL 源端准备

说实话，很多人在 Debezium 上栽跟头，不是因为连接器配置有多复杂，而是底层的环境没搭稳。Kafka 集群没调好、ZooKeeper 脑裂了、MySQL binlog 格式不对——这些问题一旦出现，排查起来特别痛苦。我早期有个项目，上线第一天连接器就断连，折腾了两天才发现是 Kafka 的 offsets.topic.replication.factor 设成了 1，单点挂了整个链路就断了。

所以这一章，咱们把地基打牢。我会带着你一步步把 Kafka 集群、ZooKeeper 和 MySQL 源端准备好。嗯，都是实战里反复验证过的套路。

整体架构概览

先看一张图，把今天要搭的东西串起来。你心里有个谱，后面操作起来就不容易乱。

ZooKeeper 集群配置

Kafka 从 2.8 开始可以脱离 ZooKeeper，但生产环境里绝大多数团队还是用 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 -C /opt/
mv /opt/apache-zookeeper-3.8.1-bin /opt/zookeeper

2. 配置文件 zoo.cfg

每个节点都要配，唯一不同的是 dataDir 和 server.X 里的 IP。我一般把数据目录单独挂一块盘，避免和系统日志抢 IO。

tickTime=2000
initLimit=10
syncLimit=5
dataDir=/data/zookeeper
clientPort=2181

server.1=192.168.1.101:2888:3888
server.2=192.168.1.102:2888:3888
server.3=192.168.1.103:2888:3888

小技巧： 每个节点上还要在 dataDir 目录下创建 myid 文件，内容就是对应的编号。比如节点 1 就写 1，别搞混了。我见过有人三个节点都写 1，结果集群起不来。

3. 启动与验证

# 每个节点依次启动
/opt/zookeeper/bin/zkServer.sh start

# 查看状态
/opt/zookeeper/bin/zkServer.sh status

# 输出示例（其中一个节点是 leader，其余是 follower）
ZooKeeper JMX enabled by default
Using config: /opt/zookeeper/bin/../conf/zoo.cfg
Client port found: 2181. Client address: localhost.
Mode: follower

看到 Mode: leader 或 follower 就对了。如果全是 standalone，说明集群没组起来，检查防火墙和 myid。

Kafka 集群搭建

Kafka 的配置项很多，但咱们只关注跟 Debezium 相关的几个核心参数。其他的保持默认就行，别过度优化。

1. 下载与解压

wget https://archive.apache.org/dist/kafka/3.4.0/kafka_2.13-3.4.0.tgz
tar -xzf kafka_2.13-3.4.0.tgz -C /opt/
mv /opt/kafka_2.13-3.4.0 /opt/kafka

2. 核心配置 server.properties

下面这个配置是我在多个生产环境验证过的。注意 offsets.topic.replication.factor 和 transaction.state.log.replication.factor 必须设成跟集群节点数一致，否则 Debezium 启动时会报错。

broker.id=1
listeners=PLAINTEXT://0.0.0.0:9092
advertised.listeners=PLAINTEXT://192.168.1.101:9092
log.dirs=/data/kafka/logs
num.partitions=3
default.replication.factor=3
offsets.topic.replication.factor=3
transaction.state.log.replication.factor=3
transaction.state.log.min.isr=2
min.insync.replicas=2
zookeeper.connect=192.168.1.101:2181,192.168.1.102:2181,192.168.1.103:2181
auto.create.topics.enable=false

注意： auto.create.topics.enable 我建议设成 false。Debezium 会自动创建 topic，但自动创建有时会使用默认的分区数和副本数，跟你的预期不一致。手动管理 topic 更可控。

3. 启动集群

# 每个节点执行
/opt/kafka/bin/kafka-server-start.sh -daemon /opt/kafka/config/server.properties

# 验证集群状态
/opt/kafka/bin/kafka-broker-api-versions.sh --bootstrap-server 192.168.1.101:9092

# 创建测试 topic
/opt/kafka/bin/kafka-topics.sh --create --topic test-topic \
  --bootstrap-server 192.168.1.101:9092 \
  --partitions 3 --replication-factor 3

用 kafka-topics.sh --describe 查看 topic 详情，确认副本都在 ISR 列表里。如果有副本不在 ISR 中，先别急着往下走，排查网络或磁盘问题。

MySQL 源端准备

MySQL 的配置是 Debezium 能否正常工作的关键。说白了，Debezium 就是靠读取 MySQL 的 binlog 来捕获变更的。所以 binlog 必须开，而且格式必须是 ROW。

1. 开启 binlog

编辑 MySQL 配置文件 /etc/my.cnf，在 [mysqld] 段下添加：

server-id=1
log_bin=mysql-bin
binlog_format=ROW
binlog_row_image=FULL
expire_logs_days=7
max_binlog_size=1G

  关键点： binlog_row_image=FULL 确保 binlog 记录变更前后的完整数据行。如果设成 MINIMAL，Debezium 拿不到旧值，某些场景下会出问题。我踩过这个坑，后来统一改成 FULL。

2. 创建 Debezium 专用用户

Debezium 需要读取 binlog 和数据库元数据。权限给最小集，别给 super 权限。

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

权限说明：

权限	用途
SELECT	读取表结构、做快照
RELOAD	执行 FLUSH 操作，用于一致性快照
SHOW DATABASES	列举数据库
REPLICATION SLAVE	连接主库读取 binlog
REPLICATION CLIENT	查看主库状态

3. 准备测试表

创建一个测试库和表，后面章节我们会用这个表来验证 Debezium 是否正常工作。

CREATE DATABASE demo;
USE demo;

CREATE TABLE users (
  id INT AUTO_INCREMENT PRIMARY KEY,
  name VARCHAR(100) NOT NULL,
  email VARCHAR(200),
  created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
) ENGINE=InnoDB;

INSERT INTO users (name, email) VALUES ('Alice', 'alice@example.com');
INSERT INTO users (name, email) VALUES ('Bob', 'bob@example.com');

验证环境连通性

所有组件都启动后，做一次快速验证。确保 Kafka 能读写，MySQL 的 binlog 能被外部访问。

# 1. 验证 Kafka 生产消费
/opt/kafka/bin/kafka-console-producer.sh --topic test-topic --bootstrap-server 192.168.1.101:9092
# 输入一条消息：hello debezium

/opt/kafka/bin/kafka-console-consumer.sh --topic test-topic --bootstrap-server 192.168.1.101:9092 --from-beginning
# 应该看到：hello debezium

# 2. 验证 MySQL binlog 状态
mysql -u debezium -p -e "SHOW MASTER STATUS;"
# 应该看到 File 和 Position 字段，说明 binlog 正在写入

避坑指南： 我曾经在验证时发现 Kafka 能生产但消费不到消息，折腾半天发现是 advertised.listeners 配成了 localhost，消费者从外部连不上。记住，advertised.listeners 一定要用其他节点能访问的 IP。

本章小结

到这一步，你的环境应该已经准备好了：

ZooKeeper 3 节点集群，正常运行
Kafka 3 节点集群，topic 副本正常
MySQL 开启了 ROW 格式 binlog，有专用用户和测试表