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

Windows 本地搭建 CDC 全链路测试环境(KafkaCanelbinlogSpringBoot)MySQL Binlog CDC 全链路实现详解:https://blog.csdn.net/BADAO_LIUMANG_QIZHI/article/details/162845686前提条件Windows 电脑MySQL 8 已安装端口 3306JDK 8 或 11 已安装java -version能执行Maven 已安装mvn -version能执行注博客https://blog.csdn.net/badao_liumang_qizhi步骤一配置 MySQL 开启 Binlog1. 找到 MySQL 配置文件通常位于C:\ProgramData\MySQL\MySQL Server 8.0\my.ini如果不确定位置在 MySQL 中执行SHOWVARIABLESLIKEdatadir;配置文件一般在 datadir 的上层目录。2. 编辑 my.ini在[mysqld]节下添加如果已有则修改[mysqld] server-id1 log-binmysql-bin binlog-formatROW binlog-row-imageFULL3. 重启 MySQLnet stop mysql80 net start mysql80服务名可能是mysql80或MySQL根据你安装时的名字4. 验证 Binlog 已开启SHOWVARIABLESLIKElog_bin;-- 结果应为 ONSHOWVARIABLESLIKEbinlog_format;-- 结果应为 ROWSHOWMASTERSTATUS;-- 应该能看到 File 和 Position5. 创建 Canal 专用用户CREATEUSERcanal%IDENTIFIEDBYcanal123;GRANTSELECT,REPLICATIONSLAVE,REPLICATIONCLIENTON*.*TOcanal%;FLUSHPRIVILEGES;6. 创建测试库和表CREATEDATABASEIFNOTEXISTScdc_demo;USEcdc_demo;CREATETABLEt_user(idINTAUTO_INCREMENTPRIMARYKEY,usernameVARCHAR(50)NOTNULL,emailVARCHAR(100),ageINT,create_timeDATETIMEDEFAULTCURRENT_TIMESTAMP,update_timeDATETIMEDEFAULTCURRENT_TIMESTAMPONUPDATECURRENT_TIMESTAMP);步骤二安装 Kafka1. 下载浏览器打开 https://kafka.apache.org/downloads下载kafka_2.13-3.6.1.tgzBinary downloads 下的 Scala 2.13 版本2. 解压解压到D:\kafka路径不要有中文和空格3. 启动 ZooKeeper打开一个CMD窗口不是 PowerShellcd /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 id0] 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。步骤三安装 Canal1. 下载浏览器打开 https://github.com/alibaba/canal/releases下载canal.deployer-1.1.7.tar.gz2. 解压解压到D:\canal3. 修改 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 335544324. 修改 instance.properties编辑D:\canal\conf\example\instance.properties# MySQL 地址 canal.instance.master.address127.0.0.1:3306 # Canal 的 slave-id不能与 MySQL 的 server-id 相同 canal.instance.mysql.slaveId12345 # 数据库账号密码 canal.instance.dbUsernamecanal canal.instance.dbPasswordcanal123 canal.instance.connectionCharsetUTF-8 # 只监听 cdc_demo 库 canal.instance.filter.regexcdc_demo\\..* # 输出到 Kafka 的 topic canal.mq.topicbinlog-cdc-demo canal.mq.partitionsNum15. 启动 Canal打开一个 CMD 窗口cd /d D:\canal\bin startup.bat如果没有startup.bat手动创建D:\canal\start.cmdecho off cd /d D:\canal java -Xms128m -Xmx512m -Dcanal.homeD:\Software\canal -Dlogback.configurationFileD:\Software\canal\conf\logback.xml -cp D:\canal\conf;D:\Software\canal\lib\* com.alibaba.otter.canal.deployer.CanalLauncher pause6. 验证 Canal 启动查看日志文件D:\canal\logs\canal\canal.logINFO 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.logINFO 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,testtest.com,20);Kafka 消费者窗口应该出现一条 JSON 消息。如果能看到消息说明 MySQL → Canal → Kafka 链路已通。步骤四创建 Spring Boot Demo 工程1. 创建项目目录在你喜欢的位置创建项目例如D:\WorkSpace\cdc-demo2. 创建 pom.xmlD:\WorkSpace\cdc-demo\pom.xml?xml version1.0 encodingUTF-8?projectxmlnshttp://maven.apache.org/POM/4.0.0xmlns:xsihttp://www.w3.org/2001/XMLSchema-instancexsi:schemaLocationhttp://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsdmodelVersion4.0.0/modelVersiongroupIdcom.example/groupIdartifactIdcdc-demo/artifactIdversion1.0.0/versionparentgroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-parent/artifactIdversion2.7.18/version/parentpropertiesjava.version1.8/java.version/propertiesdependenciesdependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-web/artifactId/dependencydependencygroupIdorg.springframework.kafka/groupIdartifactIdspring-kafka/artifactId/dependencydependencygroupIdcom.alibaba/groupIdartifactIdfastjson/artifactIdversion2.0.40/version/dependency/dependenciesbuildpluginsplugingroupIdorg.springframework.boot/groupIdartifactIdspring-boot-maven-plugin/artifactId/plugin/plugins/build/project3. 创建 application.ymlD:\WorkSpace\cdc-demo\src\main\resources\application.ymlserver: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.StringDeserializer4. 创建启动类D:\WorkSpace\cdc-demo\src\main\java\com\example\cdcdemo\CdcDemoApplication.javapackagecom.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.javapackagecom.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{privatestaticfinalLoggerlogLoggerFactory.getLogger(BinlogConsumer.class);KafkaListener(topicsbinlog-cdc-demo)publicvoidonMessage(ConsumerRecordString,Stringrecord){try{JSONObjectjsonJSON.parseObject(record.value());Stringdatabasejson.getString(database);Stringtablejson.getString(table);Stringtypejson.getString(type);JSONArraydataArrayjson.getJSONArray(data);JSONArrayoldArrayjson.getJSONArray(old);System.out.println();System.out.println( 收到 Binlog 事件 );System.out.println(库表: database.table);System.out.println(操作: type);if(dataArray!null){for(inti0;idataArray.size();i){System.out.println(数据: dataArray.getJSONObject(i).toJSONString());}}if(oldArray!null){for(inti0;ioldArray.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或者用 IDEIntelliJ/Kiro直接运行CdcDemoApplication的 main 方法。步骤五端到端验证确保所有组件已启动组件确认方式MySQLmysql -u root -p能连上ZooKeeperCMD 窗口无报错端口 2181KafkaCMD 窗口无报错端口 9092Canallogs\example\example.log中有start successfulSpring Boot控制台输出Started CdcDemoApplication执行测试在 MySQL 中逐条执行每执行一条观察 Spring Boot 控制台输出USEcdc_demo;-- 测试1INSERTINSERTINTOt_user(username,email,age)VALUES(张三,zhangsantest.com,25);-- 测试2UPDATEUPDATEt_userSETage26WHEREusername张三;-- 测试3多字段 UPDATEUPDATEt_userSETemailnewtest.com,age30WHEREusername张三;-- 测试4INSERT 多条INSERTINTOt_user(username,email,age)VALUES(李四,lisitest.com,28);-- 测试5DELETEDELETEFROMt_userWHEREusername李四;预期结果每执行一条 SQLSpring Boot 控制台应该在 1-3 秒内打印对应的 binlog 事件。测试1 预期输出 收到 Binlog 事件 库表: cdc_demo.t_user 操作: INSERT 数据: {id:1,username:张三,email:zhangsantest.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:zhangsantest.com,age:26,...} 旧值: {age:25} 测试5 预期输出 收到 Binlog 事件 库表: cdc_demo.t_user 操作: DELETE 数据: {id:2,username:李四,email:lisitest.com,age:28,...} 步骤六验证通过后清理验证完成后可以停止所有组件1. 关闭 Spring BootCtrlC 或停止 IDE 2. 停止 CanalD:\Software\canal\bin\stop.bat 3. 停止 Kafka关闭 Kafka 的 CMD 窗口CtrlC 4. 停止 ZooKeeper关闭 ZooKeeper 的 CMD 窗口CtrlC问题排查速查表现象排查步骤Canal 启动报Authentication failed检查 canal 用户密码是否正确MySQL 8 可能需要ALTER USER canal% IDENTIFIED WITH mysql_native_password BY canal123;Canal 启动报Could not find first log file删除D:\canal\conf\example\meta.dat文件后重启Kafka console-consumer 无消息检查 Canal 的canal.mq.topic是否与你监听的 topic 一致Spring Boot 无输出检查application.yml中 topic 名称是否拼写正确检查group-id是否是全新的如果之前消费过需要从头消费延迟很大超过10秒检查 Canal 日志有无异常检查 Kafka Broker 是否正常Canal 报Slave can not handle replication eventsMySQLbinlog_format不是 ROW重新配置并重启 MySQL