Windows 本地搭建 CDC 全链路测试环境(Kafka+Canel+binlog+SpringBoot)

Windows 本地搭建 CDC 全链路测试环境(Kafka+Canel+binlog+SpringBoot)

MySQL Binlog CDC 全链路实现详解:

https://blog.csdn.net/BADAO_LIUMANG_QIZHI/article/details/162845686

前提条件

• Windows 电脑
• MySQL 8 已安装（端口 3306）
• JDK 8 或 11 已安装（java -version能执行）
• Maven 已安装（mvn -version能执行）

注：

博客：

https://blog.csdn.net/badao_liumang_qizhi

步骤一：配置 MySQL 开启 Binlog

1. 找到 MySQL 配置文件

通常位于C:\ProgramData\MySQL\MySQL Server 8.0\my.ini

如果不确定位置，在 MySQL 中执行：

SHOWVARIABLESLIKE'datadir';

配置文件一般在 datadir 的上层目录。

2. 编辑 my.ini

在[mysqld]节下添加（如果已有则修改）：

[mysqld] server-id=1 log-bin=mysql-bin binlog-format=ROW binlog-row-image=FULL

3. 重启 MySQL

net stop mysql80 net start mysql80

（服务名可能是mysql80或MySQL，根据你安装时的名字）

4. 验证 Binlog 已开启

SHOWVARIABLESLIKE'log_bin';-- 结果应为 ONSHOWVARIABLESLIKE'binlog_format';-- 结果应为 ROWSHOWMASTERSTATUS;-- 应该能看到 File 和 Position

5. 创建 Canal 专用用户

CREATEUSER'canal'@'%'IDENTIFIEDBY'canal123';GRANTSELECT,REPLICATIONSLAVE,REPLICATIONCLIENTON*.*TO'canal'@'%';FLUSHPRIVILEGES;

6. 创建测试库和表

CREATEDATABASEIFNOTEXISTScdc_demo;USEcdc_demo;CREATETABLEt_user(idINTAUTO_INCREMENTPRIMARYKEY,usernameVARCHAR(50)NOTNULL,emailVARCHAR(100),ageINT,create_timeDATETIMEDEFAULTCURRENT_TIMESTAMP,update_timeDATETIMEDEFAULTCURRENT_TIMESTAMPONUPDATECURRENT_TIMESTAMP);

步骤二：安装 Kafka

1. 下载

浏览器打开 https://kafka.apache.org/downloads

下载kafka_2.13-3.6.1.tgz（Binary downloads 下的 Scala 2.13 版本）

2. 解压

解压到D:\kafka（路径不要有中文和空格）

3. 启动 ZooKeeper

打开一个CMD窗口（不是 PowerShell）：

cd /d D:\kafka bin\windows\zookeeper-server-start.bat config\zookeeper.properties

看到binding to port 0.0.0.0/0.0.0.0:2181表示成功。不要关闭此窗口。

4. 启动 Kafka Broker

再打开一个 CMD 窗口：

cd /d D:\kafka bin\windows\kafka-server-start.bat config\server.properties

看到[KafkaServer id=0] started表示成功。不要关闭此窗口。

5. 创建 Topic

再打开一个 CMD 窗口：

cd /d D:\kafka bin\windows\kafka-topics.bat --create --topic binlog-cdc-demo --bootstrap-server localhost:9092 --partitions 1 --replication-factor 1

验证：

bin\windows\kafka-topics.bat --list --bootstrap-server localhost:9092

应该能看到binlog-cdc-demo。

步骤三：安装 Canal

1. 下载

浏览器打开 https://github.com/alibaba/canal/releases

下载canal.deployer-1.1.7.tar.gz

2. 解压

解压到D:\canal

3. 修改 canal.properties

编辑D:\canal\conf\canal.properties，找到以下配置项并修改：

# 找到这一行，改为 kafka canal.serverMode = kafka # 找到 kafka 相关配置，修改为： kafka.bootstrap.servers = 127.0.0.1:9092 kafka.acks = all kafka.compression.type = none kafka.batch.size = 16384 kafka.linger.ms = 1 kafka.max.request.size = 1048576 kafka.buffer.memory = 33554432

4. 修改 instance.properties

编辑D:\canal\conf\example\instance.properties：

# MySQL 地址 canal.instance.master.address=127.0.0.1:3306 # Canal 的 slave-id（不能与 MySQL 的 server-id 相同） canal.instance.mysql.slaveId=12345 # 数据库账号密码 canal.instance.dbUsername=canal canal.instance.dbPassword=canal123 canal.instance.connectionCharset=UTF-8 # 只监听 cdc_demo 库 canal.instance.filter.regex=cdc_demo\\..* # 输出到 Kafka 的 topic canal.mq.topic=binlog-cdc-demo canal.mq.partitionsNum=1

5. 启动 Canal

打开一个 CMD 窗口：

cd /d D:\canal\bin startup.bat

如果没有startup.bat，手动创建D:\canal\start.cmd：

@echo off cd /d D:\canal java -Xms128m -Xmx512m -Dcanal.home=D:\Software\canal -Dlogback.configurationFile=D:\Software\canal\conf\logback.xml -cp "D:\canal\conf;D:\Software\canal\lib\*" com.alibaba.otter.canal.deployer.CanalLauncher pause

6. 验证 Canal 启动

查看日志文件D:\canal\logs\canal\canal.log：

INFO c.a.otter.canal.deployer.CanalLauncher - ## start the canal server INFO c.a.otter.canal.deployer.CanalController - ## start the canal server[127.0.0.1:11111]

查看D:\canal\logs\example\example.log：

INFO c.a.otter.canal.instance.core.AbstractCanalInstance - start successful...

7. 验证 Canal → Kafka 通路

打开一个 CMD 窗口，启动 Kafka 消费者监听：

cd /d D:\kafka bin\windows\kafka-console-consumer.bat --bootstrap-server localhost:9092 --topic binlog-cdc-demo --from-beginning

然后在 MySQL 中执行：

USEcdc_demo;INSERTINTOt_user(username,email,age)VALUES('test','test@test.com',20);

Kafka 消费者窗口应该出现一条 JSON 消息。如果能看到消息，说明 MySQL → Canal → Kafka 链路已通。

步骤四：创建 Spring Boot Demo 工程

1. 创建项目目录

在你喜欢的位置创建项目，例如D:\WorkSpace\cdc-demo

2. 创建 pom.xml

D:\WorkSpace\cdc-demo\pom.xml：

<?xml version="1.0" encoding="UTF-8"?><projectxmlns="http://maven.apache.org/POM/4.0.0"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"><modelVersion>4.0.0</modelVersion><groupId>com.example</groupId><artifactId>cdc-demo</artifactId><version>1.0.0</version><parent><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-parent</artifactId><version>2.7.18</version></parent><properties><java.version>1.8</java.version></properties><dependencies><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><dependency><groupId>org.springframework.kafka</groupId><artifactId>spring-kafka</artifactId></dependency><dependency><groupId>com.alibaba</groupId><artifactId>fastjson</artifactId><version>2.0.40</version></dependency></dependencies><build><plugins><plugin><groupId>org.springframework.boot</groupId><artifactId>spring-boot-maven-plugin</artifactId></plugin></plugins></build></project>

3. 创建 application.yml

D:\WorkSpace\cdc-demo\src\main\resources\application.yml：

server:port:8090spring:kafka:bootstrap-servers:127.0.0.1:9092consumer:group-id:cdc-demo-groupauto-offset-reset:earliestkey-deserializer:org.apache.kafka.common.serialization.StringDeserializervalue-deserializer:org.apache.kafka.common.serialization.StringDeserializer

4. 创建启动类

D:\WorkSpace\cdc-demo\src\main\java\com\example\cdcdemo\CdcDemoApplication.java：

packagecom.example.cdcdemo;importorg.springframework.boot.SpringApplication;importorg.springframework.boot.autoconfigure.SpringBootApplication;@SpringBootApplicationpublicclassCdcDemoApplication{publicstaticvoidmain(String[]args){SpringApplication.run(CdcDemoApplication.class,args);}}

5. 创建 Binlog 消费者

D:\WorkSpace\cdc-demo\src\main\java\com\example\cdcdemo\consumer\BinlogConsumer.java：

packagecom.example.cdcdemo.consumer;importcom.alibaba.fastjson.JSON;importcom.alibaba.fastjson.JSONArray;importcom.alibaba.fastjson.JSONObject;importorg.apache.kafka.clients.consumer.ConsumerRecord;importorg.slf4j.Logger;importorg.slf4j.LoggerFactory;importorg.springframework.kafka.annotation.KafkaListener;importorg.springframework.stereotype.Component;@ComponentpublicclassBinlogConsumer{privatestaticfinalLoggerlog=LoggerFactory.getLogger(BinlogConsumer.class);@KafkaListener(topics="binlog-cdc-demo")publicvoidonMessage(ConsumerRecord<String,String>record){try{JSONObjectjson=JSON.parseObject(record.value());Stringdatabase=json.getString("database");Stringtable=json.getString("table");Stringtype=json.getString("type");JSONArraydataArray=json.getJSONArray("data");JSONArrayoldArray=json.getJSONArray("old");System.out.println();System.out.println("========== 收到 Binlog 事件 ==========");System.out.println("库表: "+database+"."+table);System.out.println("操作: "+type);if(dataArray!=null){for(inti=0;i<dataArray.size();i++){System.out.println("数据: "+dataArray.getJSONObject(i).toJSONString());}}if(oldArray!=null){for(inti=0;i<oldArray.size();i++){System.out.println("旧值: "+oldArray.getJSONObject(i).toJSONString());}}System.out.println("=======================================");System.out.println();}catch(Exceptione){log.error("处理异常",e);}}}

6. 启动应用

在项目目录下执行：

cd /d D:\WorkSpace\cdc-demo mvn spring-boot:run

或者用 IDE（IntelliJ/Kiro）直接运行CdcDemoApplication的 main 方法。

步骤五：端到端验证

确保所有组件已启动

执行测试

在 MySQL 中逐条执行，每执行一条观察 Spring Boot 控制台输出：

USEcdc_demo;-- 测试1：INSERTINSERTINTOt_user(username,email,age)VALUES('张三','zhangsan@test.com',25);-- 测试2：UPDATEUPDATEt_userSETage=26WHEREusername='张三';-- 测试3：多字段 UPDATEUPDATEt_userSETemail='new@test.com',age=30WHEREusername='张三';-- 测试4：INSERT 多条INSERTINTOt_user(username,email,age)VALUES('李四','lisi@test.com',28);-- 测试5：DELETEDELETEFROMt_userWHEREusername='李四';

预期结果

每执行一条 SQL，Spring Boot 控制台应该在 1-3 秒内打印对应的 binlog 事件。

测试1 预期输出：

========== 收到 Binlog 事件 ========== 库表: cdc_demo.t_user 操作: INSERT 数据: {"id":"1","username":"张三","email":"zhangsan@test.com","age":"25","create_time":"2025-07-13 10:00:00","update_time":"2025-07-13 10:00:00"} =======================================

测试2 预期输出：

========== 收到 Binlog 事件 ========== 库表: cdc_demo.t_user 操作: UPDATE 数据: {"id":"1","username":"张三","email":"zhangsan@test.com","age":"26",...} 旧值: {"age":"25"} =======================================

测试5 预期输出：

========== 收到 Binlog 事件 ========== 库表: cdc_demo.t_user 操作: DELETE 数据: {"id":"2","username":"李四","email":"lisi@test.com","age":"28",...} =======================================

步骤六：验证通过后清理

验证完成后可以停止所有组件：

1. 关闭 Spring Boot（Ctrl+C 或停止 IDE） 2. 停止 Canal：D:\Software\canal\bin\stop.bat 3. 停止 Kafka：关闭 Kafka 的 CMD 窗口（Ctrl+C） 4. 停止 ZooKeeper：关闭 ZooKeeper 的 CMD 窗口（Ctrl+C）

问题排查速查表