1. 项目背景与方案选型思考做实时数据管道的人几乎都会遇到这个需求MySQL 业务表里的数据发生变化后要尽可能快地把变化同步到 Kafka让下游实时计算、数仓入仓、缓存刷新等系统去消费。以前处理这种需求要么在业务代码里双写 Kafka要么写定时任务扫表两条路都让人头疼。后来我用 Kafka-connect-jdbc-source 连接 MySQL 数据库用 JDBC Source 直接轮询数据表增量数据自动落入 Kafka topic链路干净很多。这篇博文把从环境准备、配置编写到问题排查的完整过程写下来适合刚接触 Kafka Connect 的开发者也适合正在做数据同步方案选型的人参考。先给你一个最重要的心理预期它并不是 CDC。JDBC Source 的本质是周期执行 SQL 查询靠自增主键或时间列做增量游标不是解析 binlog。它天然捕获不到 delete 事件对 update 也只能依赖时间列覆盖。理解这一点后面很多问题都不会踩。1.1 它到底解决了什么问题在没有 Kafka Connect 之前业务代码双写是很多团队的默认选择。双写的问题非常明显业务逻辑里到处散落着 Kafka 发送代码事务和消息发送的边界很难统一一旦发送失败还要补偿重试侵入性极强。定时任务扫表则更进一步虽然不侵入业务代码但实时性差扫描压力大而且增量游标需要自己维护表多了以后就是一场灾难。Kafka-connect-jdbc-source 把这些事全部封装成了一个标准连接器它负责周期性查询目标表自己维护增量游标把查询结果转换成 Kafka 消息并写入 topic。你只需要写配置不需要写代码。最直接的好处是业务系统完全无感数据库权限也可以只给只读账号管道出问题不会反向影响业务。我参与过的一个订单同步场景里数据只有 insert偶尔 update 状态没有 delete 的强需求。当时评估了几种方案最后选择 JDBC Source理由很简单不想引入额外解析组件下游只需要新增数据做指标计算update 靠时间列覆盖就能满足。事后复盘这个决策是对的。1.2 与 Canal、Debezium 等方案的取舍很多文章会拿 JDBC Source 和基于 binlog 的方案对比结论往往是“CDC 更强大”。这句话没错但脱离场景谈强大没有意义。我做了一个简洁的对比方案原理能捕获 delete 吗部署复杂度适用场景JDBC Source周期 SQL 轮询游标增量不能低一个连接器只关注新增/部分更新的场景Canal模拟 MySQL 从库解析 binlog能中需要额外部署 Canal 服务完整变更捕获强实时Debezium同样是 binlog但作为 Kafka Connect 连接器能中高配置和调优更多数据入湖、入仓需要完整历史变更这个表不是让你直接选 JDBC Source而是提醒你方案选型先回答三个问题第一业务是否需要 delete 事件第二update 是不是必须精确捕获第三团队愿不愿意多维护一套解析服务。如果三个答案都是否JDBC Source 就是性价比最高的方案。如果有一个答案是是建议认真评估 Debezium。强行用 JDBC Source 硬扛完整 CDC 需求后面一定会被语义缺失坑到。1.3 整体数据流数据流向是MySQL 表 - JDBC Source 连接器 - Kafka topic - 下游消费者。连接器周期性地对目标表执行查询把结果记录封装成 Kafka Connect 的 SourceRecord再通过 converter 序列化为 JSON 或其他格式写入 topic。Kafka Connect 的 remote 端调度模型很清晰一个 connector 负责定义数据从哪里来、往哪里去一个或多个 task 负责真正干活。JDBC Source 的 task 每次跑完一批数据会把当前游标位置提交到 offset 存储里下次启动就从上次的位置继续。这个机制保证了重启之后不会从头重新扫表但也带来了重复数据的可能后面排查章节会重点展开。2. 环境准备搭一套能跑通的最小环境纸上谈兵没有意义先把环境搭起来。这块看起来简单实际坑不少我把每一步的关键点都拆开讲。2.1 版本怎么选兼容性是第一道坑版本组合直接影响后面所有体验。我建议按照“Kafka 3.x JDK 11 或 17 MySQL 8.0 社区版 JDBC Source 插件包”这套组合来跑起来比较稳。组件版本建议说明Kafka3.4.x 或 3.5.x自带 Kafka Connect不需要额外安装JDBC Source 插件包与 Kafka 版本兼容的社区发行版主类以插件包说明为准MySQL Connector/J8.0.3x要与 MySQL 8 匹配JDK11 或 17Kafka 3.x 对两者支持都比较成熟实际操作中最难受的问题是插件包和 Kafka 版本不兼容。有些老插件包在较新的 Kafka 上会报内部 API 错误所以我建议先看插件包的 release notes确认它声明支持的 Kafka 版本范围再决定具体版本。不要盲目追求最新能稳定运行比版本新更重要。2.2 准备 MySQL 测试表和数据我先建一张订单表结构尽量贴近真实业务包含自增主键、业务单号、用户ID、金额、创建时间和更新时间。CREATE DATABASE demo DEFAULT CHARACTER SET utf8mb4; CREATE TABLE demo.order_info ( id bigint NOT NULL AUTO_INCREMENT, order_no varchar(64) NOT NULL, user_id bigint NOT NULL, amount decimal(12,2) NOT NULL, create_time datetime NOT NULL DEFAULT CURRENT_TIMESTAMP, update_time datetime NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, PRIMARY KEY (id), KEY idx_update_time (update_time) ) ENGINEInnoDB; INSERT INTO demo.order_info (order_no, user_id, amount) VALUES (ORD20240101001, 1001, 99.90), (ORD20240101002, 1002, 199.00);注意update_time我建了索引。这一点非常重要后面如果使用 timestamp 模式查询会高频使用该列做过滤条件没有索引会导致全表扫描数据量大时直接把 MySQL 拖垮。接着创建一个最小权限账号只给 SELECT 权限CREATE USER connect_user% IDENTIFIED BY your_password; GRANT SELECT ON demo.* TO connect_user%; FLUSH PRIVILEGES;只读账号是底线。连接器只需要读表不需要任何写权限给最小权限不仅是安全考虑也能避免误操作。2.3 启动 Kafka 并配置插件目录Kafka 3.3 以后可以用 KRaft 模式运行不再依赖 ZooKeeper单机实验最省事。这里给出一个最小化的config/kraft/server.propertiesprocess.rolesbroker,controller node.id1 listenersPLAINTEXT://127.0.0.1:9092,CONTROLLER://127.0.0.1:9093 advertised.listenersPLAINTEXT://127.0.0.1:9092 controller.quorum.voters1127.0.0.1:9093 log.dirs/tmp/kraft-logs启动前需要格式化存储目录格式化的 UUID 用命令生成bin/kafka-storage.sh random-uuid bin/kafka-storage.sh format -t 上面生成的uuid -c config/kraft/server.properties bin/kafka-server-start.sh config/kraft/server.propertiesKafka 起来后找一个端口确认连通正常输出启动日志后再启动 Kafka Connect。如果之前用过 ZooKeeper 模式的目录不要和 KRaft 模式混用否则会有存储状态异常。插件目录建议按下面结构单独放/opt/connect-plugins/jdbc-source/ ├── jdbc-source-connector.jar └── mysql-connector-java.jar每个连接器一个独立子目录便于管理依赖。插件目录的路径要在 Kafka Connect 的 worker 配置里指定plugin.path/opt/connect-plugins有很多人把 jar 放到任意目录改了 plugin.path 之后却没重启 Connect结果插件列表里始终看不到。记住修改插件路径或添加 jar 之后必须重启 connect 进程才会生效。2.4 验证插件被 Kafka Connect 识别启动 connect 进程后通过 REST 接口验证插件是否加载成功curl http://127.0.0.1:8083/connector-plugins返回结果里应该能看到 JDBC Source 连接器对应的 class 信息。如果看不到优先检查三类问题插件目录路径是否正确、jar 是否真的在该目录、connect 进程是否已经重启。这里最容易被忽略的是 jar 依赖冲突比如插件包自带旧版 MySQL 驱动和你放的驱动版本冲突这时候单独建子目录的方式往往能避开问题。3. 连接器核心参数搞懂原理才能排错配置不是随便抄的。JDBC Source 的参数背后是查询语义理解每个参数对应什么 SQL 行为遇到问题才不至于全靠猜。3.1 不是 CDC是查询轮询JDBC Source 不会监听 binlog它做的事情就是每隔一段时间执行一次查询。查询模式由mode控制主要有四种模式查询逻辑能捕获新增能捕获更新能捕获删除bulk每次全表查一次是但全量重复是但全量重复不保证incrementingwhere 自增列 上次最大值是否否timestamp按时间列查窗口是可能取决于时间列更新否timestampincrementing时间和自增列联合窗口是可能否bulk 模式适合小表做一次性全量同步或者定时全量快照。incrementing 模式适合只关心新增数据的场景它最简洁但 update 不会重新触发。timestamp 模式会把条件改成类似update_time 上次时间的窗口查询所以能捕获到修改了时间列的更新记录。timestampincrementing 组合模式则更稳先用时间窗口过滤再用自增列去重。这里要明确一个认知无论哪种模式它看到的都只是某一次查询时刻的表中数据不是数据库中真实发生过的变更事件。所以 delete 永远捕获不到这是架构机制决定的不是配置能解决的。3.2 增量列和时间列怎么选用 incrementing 模式时增量列必须是严格递增且非空的列最常见的当然是 BIGINT 自增主键。要注意如果业务代码会手动插入显式 ID或者主键被回拨过增量游标就会乱出现漏数据或重复数据。正常运维下 MySQL 自增主键不会回拨但如果有数据迁移或手工修数就一定要提前评估。用 timestamp 模式时时间列建议选择由数据库自动维护的update_time并加上索引。真正坑人的不是时间列本身而是两个特殊情况。第一种业务批量更新历史数据时间戳全部变成当前时间会把一大批不该同步的行重新同步一遍。第二种同一时间戳内有多行数据如果窗口边界没卡好有些行会被漏掉。组合模式能在很大程度上缓解这个问题但依然做不到精确 exactly-once。所以选型时我的经验是优先选择组合模式。增量列用主键时间列用 update_time两列都要有索引。单纯用 incrementing 虽然简单却在 update 普遍的业务里丧失了更新感知能力。3.3 配置参数逐项拆解下面是一个最常见的增量同步配置以 standalone 模式下的 properties 文件为例namemysql-order-source connector.classio.confluent.connect.jdbc.JdbcSourceConnector tasks.max1 connection.urljdbc:mysql://127.0.0.1:3306/demo?useSSLfalseserverTimezoneAsia/ShanghaiuseCursorFetchtrue connection.userconnect_user connection.passwordyour_password table.typesTABLE table.whitelistorder_info modeincrementing incrementing.column.nameid timestamp.column.nameupdate_time topic.prefixmysql-order- poll.interval.ms5000 batch.max.rows1000 table.poll.interval.ms60000 schema.patterndemo逐个解释关键参数这些是排错时绕不开的参数作用注意事项connector.class连接器主类的全限定类名不同插件发行版的类名以插件包说明为准tasks.max最大并行任务数JDBC Source 建议从 1 开始别盲目调大connection.urlJDBC 连接串必须带时区参数MySQL 8 建议开启 useCursorFetchtable.whitelist要同步的表白名单逗号分隔多张表也可以使用 table.blacklistmode轮询模式bulk、incrementing、timestamp、timestampincrementingincrementing.column.name自增列名列值必须递增timestamp.column.name时间列名建议建索引topic.prefixtopic 前缀最终 topic 名是前缀表名poll.interval.ms查询间隔毫秒数决定实时性也决定对数据库的压力batch.max.rows单批次最大行数控制每次查询返回的数据量table.poll.interval.ms表列表刷新间隔自动发现新增表的时间schema.pattern限定库名不加可能扫描所有库的表特别提醒一下topic.prefix。JDBC Source 默认生成的 topic 名是topic.prefix 表名所以上面配置会生成mysql-order-order_info这个 topic。多表同步时这个规则反而清晰每张表对应一个独立 topic。如果你希望 topic 名和表名完全一致可以考虑用 Transformation 改写 topic 名或者直接用自定义查询模式。3.4 关闭 JSON schema减小消息体积Kafka Connect 默认的 JsonConverter 会输出schema和payload两段结构schema 里带着完整字段描述每条消息都重复一遍topic 体积会大很多下游解析也更麻烦。我通常会在 worker 配置里显示关闭 schemakey.converterorg.apache.kafka.connect.json.JsonConverter value.converterorg.apache.kafka.connect.json.JsonConverter key.converter.schemas.enablefalse value.converter.schemas.enablefalse关闭后写入 Kafka 的消息就是纯粹的业务字段 JSON类似{id:1,order_no:ORD20240101001}。如果后续要用 Schema Registry 做 Avro 序列化再按需要调整本文的实战流程先用普通 JSON 足够。4. 实操两套部署模式都走一遍Kafka Connect 有两种运行模式standalone 和 distributed。我建议先跑 standalone 验证配置再切到 distributed 上生产。4.1 standalone 模式快速验证standalone 模式适合本地单机验证一个进程负责调度和执行配置简单。先写 worker 配置bootstrap.servers127.0.0.1:9092 key.converterorg.apache.kafka.connect.json.JsonConverter value.converterorg.apache.kafka.connect.json.JsonConverter key.converter.schemas.enablefalse value.converter.schemas.enablefalse offset.storage.file.filename/tmp/connect.offsets plugin.path/opt/connect-plugins再把刚才的 mysql-source.properties 也准备好然后执行bin/connect-standalone.sh config/connect-standalone.properties config/mysql-source.properties启动成功的话日志里会出现连接器创建成功、task 开始运行的信息。standalone 模式的 offset 保存在本地文件/tmp/connect.offsets连接器每消费一批数据就会更新这个文件。如果删掉这个文件下次启动就会重新从头同步这一点要特别注意。standalone 模式下连接器的创建由命令行配置文件决定不通过 REST API。所以如果你想快速验证一组参数是否合理用 standalone 是最合适的。4.2 distributed 模式创建连接器生产环境推荐使用 distributed 模式它的核心价值是支持多 worker 横向扩展和故障转移。worker 配置里需要指定三个内部 topicbootstrap.servers127.0.0.1:9092 group.idconnect-cluster-group key.converterorg.apache.kafka.connect.json.JsonConverter value.converterorg.apache.kafka.connect.json.JsonConverter key.converter.schemas.enablefalse value.converter.schemas.enablefalse config.storage.topicconnect-configs offset.storage.topicconnect-offsets status.storage.topicconnect-status plugin.path/opt/connect-plugins这三个内部 topic 是分布式模式正常运行的基础。建议手动创建分区数设置为 1复制因子根据集群环境而定。单机实验用复制因子 1 即可bin/kafka-topics.sh --bootstrap-server 127.0.0.1:9092 --create --topic connect-configs --partitions 1 --replication-factor 1 bin/kafka-topics.sh --bootstrap-server 127.0.0.1:9092 --create --topic connect-offsets --partitions 1 --replication-factor 1 bin/kafka-topics.sh --bootstrap-server 127.0.0.1:9092 --create --topic connect-status --partitions 1 --replication-factor 1启动分布式模式bin/connect-distributed.sh config/connect-distributed.properties启动成功后再通过 REST API 提交连接器curl -X POST http://127.0.0.1:8083/connectors \ -H Content-Type: application/json \ -d { name: mysql-order-source, config: { connector.class: io.confluent.connect.jdbc.JdbcSourceConnector, tasks.max: 1, connection.url: jdbc:mysql://127.0.0.1:3306/demo?useSSLfalseserverTimezoneAsia/ShanghaiuseCursorFetchtrue, connection.user: connect_user, connection.password: your_password, table.whitelist: order_info, mode: incrementing, incrementing.column.name: id, timestamp.column.name: update_time, topic.prefix: mysql-order-, poll.interval.ms: 5000, batch.max.rows: 1000 } }注意distributed 模式通过 REST API 提交连接器时所有配置项的值都是字符串。返回 201 或 200 状态码代表创建成功。之后你可以通过 GET 接口查看状态也可以随时修改配置Kafka Connect 会自动处理版本变更。4.3 验证消费与增量效果连接器一旦创建成功就会开始工作。先用 console consumer 消费目标 topicbin/kafka-console-consumer.sh \ --bootstrap-server 127.0.0.1:9092 \ --topic mysql-order-order_info \ --from-beginning因为这是第一次启动连接器会把表里的已有数据全部读出来你会看到类似下面的 JSON{id:1,order_no:ORD20240101001,user_id:1001,amount:99.90,create_time:2024-01-01 10:00:00,update_time:2024-01-01 10:00:00} {id:2,order_no:ORD20240101002,user_id:1002,amount:199.00,create_time:2024-01-01 10:05:00,update_time:2024-01-01 10:05:00}接着向 MySQL 插入一条新数据INSERT INTO demo.order_info (order_no, user_id, amount) VALUES (ORD20240101003, 1003, 299.00);等一个 poll.interval 周期consumer 端会立即出现新记录。这就是 incrementing 模式的基本效果只按 id 增量读取。再试试 update 和 delete。执行UPDATE demo.order_info SET amount 99.00 WHERE id 1; DELETE FROM demo.order_info WHERE id 2;你会发现 consumer 端什么都不会输出。这再次印证了 JDBC Source 的语义局限update 没有触发时间列变化时捕获不到delete 完全捕获不到。如果业务需要这两类事件就得换 Debezium 这类 binlog 方案。4.4 连接器状态管理与重启分布式模式下最常用的管理接口有三个# 查看所有已创建的连接器 curl http://127.0.0.1:8083/connectors # 查看某个连接器及 task 的运行状态 curl http://127.0.0.1:8083/connectors/mysql-order-source/status # 重启某个失败的 task curl -X POST http://127.0.0.1:8083/connectors/mysql-order-source/tasks/0/restart状态接口返回的 JSON 里会明确标记 RUNNING 或 FAILED。很多人在任务失败后只会删掉重建连接器其实先看状态、再重启对应 task 会更安全也能避免 offset 混乱。offset 在分布式模式下存放在connect-offsets内部 topic 里而不是本地文件。所以不要随便清理这个 topic否则连接器会丢失断点重新全量读一遍。5. 常见问题与排错心得实战中遇到的问题远比文档里写的来得多。下面这几类是我踩过坑之后觉得最值得记录的。5.1 插件加载失败与驱动缺失现象是日志里出现ClassNotFoundException或者 REST 接口返回No such connector class。常见原因有三个第一plugin.path 配置不正确。确认 connect 进程读取到的配置是不是你修改后的那个文件有些启动脚本会覆盖环境变量。第二缺少 MySQL JDBC 驱动。连接器确实能加载但执行 JDBC 操作时发现找不到驱动类报No suitable driver。这时候把 mysql-connector-java 的 jar 放进同一个插件子目录即可。第三依赖冲突。插件包本身带了旧版驱动或旧版日志库和 MySQL 驱动冲突。我的经验是把每个连接器单独放一个子目录不要一股脑把所有 jar 堆在同一个目录下。Kafka Connect 加载插件时子目录隔离能很大程度避免 classpath 污染。5.2 连接 MySQL 时常见的时区、SSL 和认证问题这个问题出现频率极高。内存报错通常有两种一种是The server time zone value ... is unrecognized另一种是Public Key Retrieval is not allowed。前者解决方法是连接串显式指定时区jdbc:mysql://127.0.0.1:3306/demo?useSSLfalseserverTimezoneAsia/Shanghai后者通常出现在 MySQL 8 使用默认认证插件caching_sha2_password时测试环境可以临时加allowPublicKeyRetrievaltruejdbc:mysql://127.0.0.1:3306/demo?useSSLfalseserverTimezoneAsia/ShanghaiallowPublicKeyRetrievaltrue生产环境不要因为这个提示就无脑打开 allowPublicKeyRetrieval应该优先用 SSL 连接或调整认证方式。另外如果密码里有特殊字符properties 文件里的 password 值要留意转义问题建议先测试连接串能否用命令行工具连通。5.3 增量模式漏数据与重复数据这是 JDBC Source 最核心的痛点也是你一定会遇到的问题。只配 incrementing 模式时如果业务对一个不改变主键的旧行做更新新值不会被捕获。这是机制决定的不算 bug但产品经理不一定认。搭配 timestamp 模式后update_time 自动更新会让该行重新出现在时间窗口里于是捕获到了。但 timestamp 模式有边界问题。举个例子假设本次查询窗口是update_time 10:00:00查询执行的同时事务 A 插入了一行 update_time 为 10:00:00。事务 A 在查询执行后才提交那么本次查询看不到它。下次窗口变成update_time 10:00:00又会把它读出来。这种边界场景会导致重复或延迟取决于具体执行时序。组合模式能减少重复但不能根治。所以如果下游对重复数据敏感一定要做幂等处理。消息里带上自增主键或业务唯一键作为去重依据这是最简单可靠的办法。不要在源端追求 exactly-onceKafka Connect 的 JDBC Source 默认语义是 at least once。5.4 大表全量导出性能与连接参数调优第一次启动时如果表里数据量很大连接器会一次性把整表数据读出来。这里有两个调节手段。第一连接串加useCursorFetchtrue配合defaultFetchSize让 MySQL 驱动使用流式读取而不是一次性把结果集全部加载到内存。示例jdbc:mysql://127.0.0.1:3306/demo?useSSLfalseserverTimezoneAsia/ShanghaiuseCursorFetchtruedefaultFetchSize1000第二调整batch.max.rows和poll.interval.ms。batch.max.rows控制的是每次轮询查询返回的最大行数吗实际上它更多影响的是分页读取行为。轮询间隔越长单位时间内对数据库的查询次数越少但实时性会下降。通常先用默认值跑再通过监控数据库慢日志判断是否需要调整。5.5 任务状态 FAILED 的通用排查路径当连接器状态变成 FAILED不要着急删掉重建按这个路径查先看状态接口curl http://127.0.0.1:8083/connectors/mysql-order-source/status然后看 worker 日志里对应的 stack trace。大部分失败原因都能在异常堆栈里定位比如连接超时、表不存在、权限不足、SQL 语法错误。如果异常信息和数据相关比如某一行数据转换失败可以考虑把tasks.max暂时改成 1减少并发干扰。修复配置后用重启 task 接口让任务重新拉起。如果反复失败再考虑删除连接器同时清理内部 topic 中对应连接器的配置和 offset。6. 落地之后的一些心得体会这个方案我已经在几个项目里实际用过了最后分享几条不一定写在官方文档里的体会。第一条连接器配置尽量模板化。不同业务表的核心差异基本就是连接串、表名、增量列和 topic 前缀把这几个参数抽成模板新接入一张表只需要填一份配置效率会高很多。不要每次都从零手写。第二条正式上生产前一定先用 standalone 模式跑一段时间观察。standalone 虽然不适合生产但它能在单机上快速证明数据链路和配置是对的也能方便地查看日志。确认无误后再迁移到 distributed。第三条监控必须覆盖三个点连接器状态是否 RUNNING、offset 是否持续推进、目标 topic 的消费水位是否正常。三个点里任意一个出问题数据链路都有隐患。不要只盯着 Kafka 集群本身的监控。第四条如果要同步的表存在频繁 update 或需要 delete 事件JDBC Source 真的不适合。与其硬上不如早换 Debezium。我们不是要证明某个工具最好而是要在正确的场景里用正确的工具。个人觉得JDBC Source 最大的价值不是技术高深而是用极低的成本解决了一条常见管道的数据同步问题。在合适的场景下它能让团队少写很多重复代码少维护很多临时脚本。理解了它的机制和边界这套实战经验能一直复用下去。