环境准备：Flink集群搭建（Standalone模式）、MySQL数据库安装与配置、Kafka集群搭建、Flink CDC依赖引入

说实话，做实时数据同步这件事，环境准备是最容易让人翻车的地方。我记得刚带团队做第一个CDC项目时，大家花了两天搭环境，结果Kafka和Flink的版本不兼容，又重来一遍。所以这一章，我把每一步的关键细节都给你拆开讲。

2.1 Flink集群搭建（Standalone模式）

Flink支持多种部署模式，Standalone模式最适合学习和开发测试。你想想看，不用依赖YARN或Kubernetes，一台机器就能跑起来，多方便。

2.1.1 下载与安装

我个人习惯用1.17.x版本，稳定且社区支持好。直接去官网下载即可。

# 下载Flink
wget https://dlcdn.apache.org/flink/flink-1.17.2/flink-1.17.2-bin-scala_2.12.tgz

# 解压
tar -xzf flink-1.17.2-bin-scala_2.12.tgz
mv flink-1.17.2 /usr/local/flink

2.1.2 配置与启动

核心配置文件就两个：flink-conf.yaml 和 workers（旧版叫 slaves）。

# flink-conf.yaml 核心配置
jobmanager.rpc.address: 192.168.1.100
jobmanager.rpc.port: 6123
jobmanager.memory.process.size: 2048m
taskmanager.memory.process.size: 2048m
taskmanager.numberOfTaskSlots: 4
parallelism.default: 2

配置好之后，启动集群：

# 启动集群
cd /usr/local/flink
./bin/start-cluster.sh

# 检查进程
jps
# 应该看到：StandaloneSessionClusterEntrypoint（JobManager）
#           TaskManagerRunner（TaskManager）

2.1.3 验证集群

跑一个简单的WordCount任务试试：

# 提交测试任务
./bin/flink run ./examples/batch/WordCount.jar

# 查看任务状态
./bin/flink list

如果任务成功运行，说明集群没问题。嗯，这里要注意，Standalone模式默认不开启HA，生产环境建议用YARN或K8s模式。

2.2 MySQL数据库安装与配置

MySQL是CDC的数据源端，我们用它来模拟业务数据库。我建议用MySQL 8.0，因为Flink CDC对8.0的支持最好。

2.2.1 安装MySQL

# Ubuntu/Debian
apt-get install mysql-server-8.0

# CentOS/RHEL
yum install mysql-server

# 启动服务
systemctl start mysqld
systemctl enable mysqld

2.2.2 开启Binlog

CDC的核心依赖Binlog，所以必须开启。我曾经遇到过项目上线后才发现Binlog没开，数据同步一直没反应，排查了半天。

# 编辑MySQL配置文件
vim /etc/mysql/mysql.conf.d/mysqld.cnf

# 添加以下配置
[mysqld]
server-id = 1
log_bin = mysql-bin
binlog_format = ROW
binlog_row_image = FULL
expire_logs_days = 7

配置完成后重启MySQL：

systemctl restart mysqld

# 验证Binlog是否开启
mysql -u root -p -e "SHOW VARIABLES LIKE 'log_bin';"
# 结果应为：ON

2.2.3 创建CDC用户

Flink CDC需要专门的数据库用户，权限要够但别太大：

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

你想想看，REPLICATION SLAVE 权限是读Binlog必需的，REPLICATION CLIENT 用来获取Binlog位置信息。少了任何一个，Flink CDC都连不上。

2.3 Kafka集群搭建

Kafka在CDC架构中扮演消息队列的角色，Flink把数据写到Kafka，下游消费者再拉取。说白了，它就是数据管道的。

2.3.1 下载与安装

# 下载Kafka（自带ZooKeeper）
wget https://dlcdn.apache.org/kafka/3.5.1/kafka_2.12-3.5.1.tgz
tar -xzf kafka_2.12-3.5.1.tgz
mv kafka_2.12-3.5.1 /usr/local/kafka

2.3.2 配置ZooKeeper

Kafka依赖ZooKeeper做协调服务。单机测试直接用内置的就行：

# 配置ZooKeeper
vim /usr/local/kafka/config/zookeeper.properties

# 核心配置
dataDir=/tmp/zookeeper
clientPort=2181
maxClientCnxns=60

2.3.3 配置Kafka Broker

vim /usr/local/kafka/config/server.properties

# 核心配置
broker.id=0
listeners=PLAINTEXT://localhost:9092
log.dirs=/tmp/kafka-logs
num.partitions=3
offsets.topic.replication.factor=1

2.3.4 启动Kafka集群

# 先启动ZooKeeper
cd /usr/local/kafka
./bin/zookeeper-server-start.sh config/zookeeper.properties &

# 再启动Kafka
./bin/kafka-server-start.sh config/server.properties &

# 验证
./bin/kafka-topics.sh --list --bootstrap-server localhost:9092

如果没报错，说明Kafka跑起来了。创建一个测试Topic确认一下：

./bin/kafka-topics.sh --create --topic test-topic \
  --bootstrap-server localhost:9092 \
  --partitions 3 --replication-factor 1

2.4 Flink CDC依赖引入

环境搭好了，接下来就是让Flink能识别CDC。说白了，就是加几个Maven依赖。

2.4.1 Maven项目配置

我建议用Maven管理项目，依赖版本要统一，否则容易冲突。

<properties>
    <flink.version>1.17.2</flink.version>
    <flink.cdc.version>2.4.2</flink.cdc.version>
    <scala.binary.version>2.12</scala.binary.version>
</properties>

<dependencies>
    <!-- Flink核心依赖 -->
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-streaming-java</artifactId>
        <version>${flink.version}</version>
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-clients</artifactId>
        <version>${flink.version}</version>
    </dependency>

    <!-- Flink CDC依赖 -->
    <dependency>
        <groupId>com.ververica</groupId>
        <artifactId>flink-connector-mysql-cdc</artifactId>
        <version>${flink.cdc.version}</version>
    </dependency>

    <!-- Kafka连接器 -->
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-connector-kafka</artifactId>
        <version>${flink.version}</version>
    </dependency>
</dependencies>

2.4.2 验证依赖是否生效

写一个简单的测试类，看看能不能加载CDC连接器：

import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class CDCTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        
        MySqlSource<String> source = MySqlSource.<String>builder()
            .hostname("localhost")
            .port(3306)
            .databaseList("test_db")
            .tableList("test_db.users")
            .username("cdc_user")
            .password("cdc_password")
            .build();

        env.fromSource(source, WatermarkStrategy.noWatermarks(), "MySQL Source")
            .print();
        
        env.execute("CDC Test");
    }
}

如果编译通过，说明依赖没问题。嗯，这里要注意，第一次跑可能会报缺少某些类，那是因为Flink CDC依赖了Jackson等库，Maven会自动下载。

2.5 整体架构图

下面这张图展示了我们搭建好的环境之间的关系。你一看就明白了：MySQL产生数据，Flink CDC读取Binlog，然后写入Kafka，整个过程是实时的。

2.6 环境验证清单

所有组件都装好后，我建议按这个清单逐一验证，避免遗漏：

序号	验证项	验证方法	预期结果
1	Flink集群	访问 http://localhost:8081	Web UI正常显示，TaskManager在线
2	MySQL Binlog	SHOW VARIABLES LIKE 'log_bin'	值为 ON
3	CDC用户权限	用cdc_user登录，执行 SHOW MASTER STATUS	显示Binlog文件名和位置
4	Kafka集群	kafka-topics.sh --list	列出已有Topic，无报错
5	Flink CDC依赖	编译运行CDCTest.java	无ClassNotFoundException