企业官网建设流程全解析

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 更强大”。这句话没错，但脱离场景谈强大没有意义。我做了一个简洁的对比：

这个表不是让你直接选 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 插件包”这套组合来，跑起来比较稳。

实际操作中，最难受的问题是插件包和 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`) ) ENGINE=InnoDB; 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.properties：

process.roles=broker,controller node.id=1 listeners=PLAINTEXT://127.0.0.1:9092,CONTROLLER://127.0.0.1:9093 advertised.listeners=PLAINTEXT://127.0.0.1:9092 controller.quorum.voters=1@127.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.properties

Kafka 起来后，找一个端口确认连通，正常输出启动日志后，再启动 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 模式适合小表做一次性全量同步，或者定时全量快照。incrementing 模式适合只关心新增数据的场景，它最简洁，但 update 不会重新触发。timestamp 模式会把条件改成类似update_time >= 上次时间的窗口查询，所以能捕获到修改了时间列的更新记录。timestamp+incrementing 组合模式则更稳，先用时间窗口过滤，再用自增列去重。

这里要明确一个认知：无论哪种模式，它看到的都只是某一次查询时刻的表中数据，不是数据库中真实发生过的变更事件。所以 delete 永远捕获不到，这是架构机制决定的，不是配置能解决的。

3.2 增量列和时间列怎么选

用 incrementing 模式时，增量列必须是严格递增且非空的列，最常见的当然是 BIGINT 自增主键。要注意，如果业务代码会手动插入显式 ID，或者主键被回拨过，增量游标就会乱，出现漏数据或重复数据。正常运维下 MySQL 自增主键不会回拨，但如果有数据迁移或手工修数，就一定要提前评估。

用 timestamp 模式时，时间列建议选择由数据库自动维护的update_time，并加上索引。真正坑人的不是时间列本身，而是两个特殊情况。第一种，业务批量更新历史数据，时间戳全部变成当前时间，会把一大批不该同步的行重新同步一遍。第二种，同一时间戳内有多行数据，如果窗口边界没卡好，有些行会被漏掉。组合模式能在很大程度上缓解这个问题，但依然做不到精确 exactly-once。

所以选型时我的经验是：优先选择组合模式。增量列用主键，时间列用 update_time，两列都要有索引。单纯用 incrementing 虽然简单，却在 update 普遍的业务里丧失了更新感知能力。

3.3 配置参数逐项拆解

下面是一个最常见的增量同步配置，以 standalone 模式下的 properties 文件为例：

name=mysql-order-source connector.class=io.confluent.connect.jdbc.JdbcSourceConnector tasks.max=1 connection.url=jdbc:mysql://127.0.0.1:3306/demo?useSSL=false&serverTimezone=Asia/Shanghai&useCursorFetch=true connection.user=connect_user connection.password=your_password table.types=TABLE 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 table.poll.interval.ms=60000 schema.pattern=demo

逐个解释关键参数，这些是排错时绕不开的：

特别提醒一下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 配置里显示关闭 schema：

key.converter=org.apache.kafka.connect.json.JsonConverter value.converter=org.apache.kafka.connect.json.JsonConverter key.converter.schemas.enable=false value.converter.schemas.enable=false

关闭后，写入 Kafka 的消息就是纯粹的业务字段 JSON，类似{"id":1,"order_no":"ORD20240101001"}。如果后续要用 Schema Registry 做 Avro 序列化，再按需要调整，本文的实战流程先用普通 JSON 足够。

4. 实操：两套部署模式都走一遍

Kafka Connect 有两种运行模式：standalone 和 distributed。我建议先跑 standalone 验证配置，再切到 distributed 上生产。

4.1 standalone 模式快速验证

standalone 模式适合本地单机验证，一个进程负责调度和执行，配置简单。先写 worker 配置：

bootstrap.servers=127.0.0.1:9092 key.converter=org.apache.kafka.connect.json.JsonConverter value.converter=org.apache.kafka.connect.json.JsonConverter key.converter.schemas.enable=false value.converter.schemas.enable=false 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 配置里需要指定三个内部 topic：

bootstrap.servers=127.0.0.1:9092 group.id=connect-cluster-group key.converter=org.apache.kafka.connect.json.JsonConverter value.converter=org.apache.kafka.connect.json.JsonConverter key.converter.schemas.enable=false value.converter.schemas.enable=false config.storage.topic=connect-configs offset.storage.topic=connect-offsets status.storage.topic=connect-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?useSSL=false&serverTimezone=Asia/Shanghai&useCursorFetch=true", "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 消费目标 topic：

bin/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?useSSL=false&serverTimezone=Asia/Shanghai

后者通常出现在 MySQL 8 使用默认认证插件caching_sha2_password时，测试环境可以临时加allowPublicKeyRetrieval=true：

jdbc:mysql://127.0.0.1:3306/demo?useSSL=false&serverTimezone=Asia/Shanghai&allowPublicKeyRetrieval=true

生产环境不要因为这个提示就无脑打开 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-once，Kafka Connect 的 JDBC Source 默认语义是 at least once。

5.4 大表全量导出性能与连接参数调优

第一次启动时，如果表里数据量很大，连接器会一次性把整表数据读出来。这里有两个调节手段。

第一，连接串加useCursorFetch=true，配合defaultFetchSize让 MySQL 驱动使用流式读取，而不是一次性把结果集全部加载到内存。示例：

jdbc:mysql://127.0.0.1:3306/demo?useSSL=false&serverTimezone=Asia/Shanghai&useCursorFetch=true&defaultFetchSize=1000

第二，调整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 最大的价值不是技术高深，而是用极低的成本解决了一条常见管道的数据同步问题。在合适的场景下，它能让团队少写很多重复代码，少维护很多临时脚本。理解了它的机制和边界，这套实战经验能一直复用下去。