开源数据同步中间件实战：从MySQL到Kafka的增量同步与避坑指南

简介DBSyncer简称dbs是一款开源的数据同步中间件面向需要跨数据库与消息系统做数据流转的开发者与运维人员解决异构数据源之间全量、增量同步及转换的业务问题。其同步场景覆盖MySQL、Oracle、SqlServer、PostgreSQL、Elasticsearch、Kafka、File与SQL等并支持上传插件自定义同步转换逻辑同时提供全量与增量数据统计图、应用性能预警等监控能力适合有一定数据库基础、需要搭建同步链路的初中级工程师参考。资源包共737个文件以472个java源码为核心辅以html、css、js、png等前端与界面资源另有xml、sql、json、sh、bat等配置与脚本文件整体约2.07MB结构完整便于二次开发。目前已有747人学习下载可据此了解同步中间件的模块划分、插件机制与监控实现思路。1. 一款开源数据同步中间件到底在解决什么脏活累活线上跑着 MySQL 的业务库报表侧是 Oracle日志落到 Kafka历史归档在 SqlServer运营偶尔还要导一份 CSV 给财务——数据散在六种存储里每加一条同步链路就要写一套定时脚本、处理一次字段类型对不上的问题。这就是数据同步中间件要接的活把「从 A 读、往 B 写」这件事抽象成配置而不是每次重写代码。它面向的是被多源异构同步折磨过的后端、数据平台和运维同学尤其是那些既不想上重型商业 ETL、又受够了 crontab 加 Python 脚本的人。开源方案的价值在于你能看到每条记录怎么被抽取、转换、装载出问题时有日志可查而不是面对一个黑匣子干瞪眼。这篇就按「它是什么、怎么搭、参数怎么调、坑在哪」的顺序把这类中间件讲透。2. 拆开看Reader、Writer 与 Channel 的三段式模型2.1 为什么是插件化架构而不是写死链路多源同步最容易踩的坑是把「读 MySQL」和「写 Kafka」的代码耦合在一个脚本里。一旦要加 Oracle 源就得复制粘贴再改一遍。成熟的开源同步中间件普遍采用 Reader / Writer / Channel 三段式Reader 负责从源端拉数据并转成内部统一记录格式Channel 负责在内存或磁盘上缓冲、限流、做脏数据隔离Writer 负责把记录按目标端方言写出去。这样 MySQL 到 Oracle、Oracle 到 Kafka、File 到 SqlServer 都只是换 Reader 和 Writer 的组合同步逻辑本身不动。选型时要盯住三点Reader 是否支持增量位点binlog、SCN、offsetChannel 是否有背压机制Writer 是否支持批量提交和事务。只支持全量、没有位点记录的方案数据量一上来就会翻车。2.2 内部记录格式决定了字段映射的难度中间件内部一般用一张扁平的 record 结构承载数据字段名、类型、值分开存。源端 MySQL 的datetime、Oracle 的DATE、Kafka 里的字符串时间戳进到内部都要归一化成统一的时间表示否则 Writer 侧没法处理。下面是一段典型的字段映射配置用 JSON 描述源字段到目标字段的转换{ source: { type: mysql, table: orders, columns: [id, user_id, amount, created_at] }, transform: [ {column: created_at, from: datetime, to: timestamp_ms}, {column: amount, from: decimal(10,2), to: string} ], sink: { type: kafka, topic: order_events, keyColumn: id } }这段配置的逻辑是从orders表读四列把created_at从数据库时间转成毫秒时间戳把amount转成字符串避免浮点精度丢失最后以id作为 Kafka 消息 key 写入order_events。参数说明上keyColumn决定分区顺序同一订单的事件会落到同一分区保证消费端顺序transform数组按顺序执行前一步的输出是后一步的输入写反了类型就对不上。2.3 增量同步的位点管理是命门全量同步谁都会写真正难的是增量。MySQL 靠 binlog 位点file position 或 GTIDOracle 靠 SCNKafka 靠 offsetFile 靠行号或修改时间。中间件必须把位点持久化到一张状态表或本地文件重启后从上次位置继续。常见做法是每批提交成功后更新位点批大小和位点更新频率要匹配——批太大重启后重复数据多批太小位点写太频繁拖慢吞吐。提示位点存储不要和目标库放同一个实例否则目标库故障时位点一起丢恢复时无从判断从哪续。3. 从零搭一条 MySQL 到 Kafka 的同步链路3.1 环境准备与依赖检查先确认源端 MySQL 开了 binlog 且格式为 ROW这是增量同步的前提。执行下面这条命令查看SHOW VARIABLES LIKE log_bin; SHOW VARIABLES LIKE binlog_format; SHOW VARIABLES LIKE binlog_row_image;期望结果是log_binON、binlog_formatROW、binlog_row_imageFULL。如果binlog_format是 STATEMENT同步中间件拿不到行级变更前后的完整值更新和删除操作会丢字段。改完配置要重启 MySQL生产环境记得在低峰期做。Kafka 侧确认 topic 已创建、分区数合理。分区数建议按目标消费并行度设置一般取消费者线程数的整数倍。用下面命令建 topickafka-topics.sh --create \ --bootstrap-server localhost:9092 \ --topic order_events \ --partitions 6 \ --replication-factor 2partitions 6意味着最多 6 个消费者并行消费replication-factor 2保证一个 broker 挂掉数据不丢。单机测试可以设 1生产至少 2。3.2 中间件配置文件的四个必调参数同步中间件的配置文件通常分 source、sink、channel 三段。以一条 MySQL 到 Kafka 的链路为例关键参数如下表参数作用建议值调错后果batchSize每批读取记录数500~2000太大内存涨太小吞吐低flushInterval强制刷出间隔(ms)1000太大延迟高太小批次碎channelCapacity内存缓冲上限10000太小背压频繁太大 OOMretryTimes写入失败重试次数3太少丢数据太多阻塞配置示例source: type: mysql host: 127.0.0.1 port: 3306 database: shop table: orders mode: incremental binlogFile: mysql-bin.000003 binlogPosition: 154 channel: type: memory capacity: 10000 batchSize: 1000 flushInterval: 1000 sink: type: kafka bootstrapServers: localhost:9092 topic: order_events acks: all retryTimes: 3mode: incremental配合binlogFile和binlogPosition指定起始位点首次全量后中间件会自动记录新位点。acks: all要求 Kafka 所有副本确认才算写入成功配合retryTimes保证不丢代价是延迟略高。3.3 启动、验证与位点确认启动中间件后往源表插一条数据观察 Kafka 是否收到kafka-console-consumer.sh \ --bootstrap-server localhost:9092 \ --topic order_events \ --from-beginning \ --max-messages 1能打印出刚插入的订单 JSON 就说明链路通了。接着查中间件的状态表确认位点已推进SELECT job_name, binlog_file, binlog_position, updated_at FROM sync_position WHERE job_name mysql2kafka_orders;binlog_position应该大于启动时配置的值updated_at是最近一次提交时间。如果位点不动先看中间件日志有没有写入报错再看 Kafka 是否可达。这一步是排查同步停滞的第一现场别跳过。4. 多源适配Oracle、SqlServer、File 各自的脾气4.1 Oracle 的 SCN 与大小写陷阱Oracle 增量同步靠 SCNSystem Change Number比 MySQL 的 binlog 位点更抽象。中间件一般通过 LogMiner 或触发器捕获变更配置时要指定起始 SCN。Oracle 默认把未加引号的标识符转成大写而 MySQL 习惯小写字段映射时user_id和USER_ID会被当成两个字段。常见做法是在 Reader 侧统一转小写或在映射配置里显式声明大小写策略。source: type: oracle host: 10.0.0.5 port: 1521 serviceName: ORCL schema: SHOP table: ORDERS mode: incremental startScn: 12345678 caseSensitive: falsecaseSensitive: false让中间件忽略大小写差异startScn指定从哪个系统变更号开始读。SCN 获取可以用SELECT CURRENT_SCN FROM V$DATABASE但生产库上频繁查这个视图有性能影响建议只在初始化时取一次。4.2 SqlServer 的 CDC 依赖与权限SqlServer 走 CDCChange Data Capture时必须先对库和表开启 CDC且执行账号要有sysadmin或db_owner角色。开启命令EXEC sys.sp_cdc_enable_db; EXEC sys.sp_cdc_enable_table source_schema dbo, source_name orders, role_name NULL;role_name NULL表示不限制读取角色测试方便但生产建议指定具体角色。CDC 表会额外占用空间cdc.lsn_time_mapping和cdc.dbo_orders_CT两张表要纳入监控否则日志涨满磁盘是迟早的事。4.3 File 与 Kafka 作为源端的特殊处理File 源常见于 CSV、JSON 行文本难点在编码和分隔符。UTF-8 带 BOM 的文件第一列字段名会多出不可见字符导致映射失败。读取时指定encoding: utf-8-sig可以吃掉 BOM。Kafka 作为源端时中间件消费的是消息位点就是 offset配置里要写group.id和auto.offset.resetsource: type: kafka bootstrapServers: localhost:9092 topic: raw_events groupId: sync_consumer_group autoOffsetReset: earliestautoOffsetReset: earliest表示没有已提交位点时从最早消息读latest则只读新消息。首次同步历史数据用earliest实时管道用latest选错会漏数据或多读一堆旧消息。5. 避坑与排查那些让同步任务半夜报警的细节5.1 现象同步延迟持续上涨位点几乎不动原因通常是 Writer 侧写入慢Channel 缓冲被填满后触发背压Reader 被迫降速。Kafka 写入慢可能是acks: all加副本同步慢或分区数太少导致单分区成为瓶颈。解决先看 Kafka broker 的磁盘 IO 和网络再考虑把acks降到1接受极小概率丢数据换吞吐或增加分区数。数据库写入慢则检查目标表有没有过多索引、是否逐条提交。5.2 现象目标端出现重复数据原因是位点更新和批次提交不在同一事务里中间件在提交后、更新位点前崩溃重启后从旧位点重读。解决Writer 侧做幂等用主键 upsert 代替 insert或把位点更新和写入放进同一事务同库时可行。跨库场景下幂等是唯一可靠方案。5.3 现象Oracle 源报 ORA-01555 快照过旧长事务或 LogMiner 读取的 undo 数据被覆盖。原因是同步任务读取慢undo 表空间回收了它需要的旧版本数据。解决加大 undo 表空间和undo_retention或缩短单批读取时间、提高同步频率。根本办法是让同步任务别落后主库太多。5.4 现象字段类型转换后精度丢失MySQLdecimal(18,4)同步到 Kafka 再进 OracleNUMBER时如果中间经过浮点类型小数位会漂移。解决金额类字段全程用字符串或定点数传递中间件 transform 里显式声明to: string目标端再转回数值。别信默认类型推断。5.5 现象Kafka 消息 key 为 null 导致分区乱序配置里漏了keyColumn或该列值为空。同一业务实体的消息散落到不同分区消费端无法保证顺序。解决选一个非空且分布均匀的列作 key比如订单号、用户 ID确实没有合适列时用源表主键拼接。6. 进阶用位点回放做数据校验与断点续传同步链路跑起来只是开始真正让人睡得着觉的是「能验证、能回放」。我一般会加一个校验任务按时间窗口从源端和目标端各拉一批数据比对记录数和关键字段的 checksum。发现不一致时用中间件记录的位点做定向回放——把位点回拨到不一致窗口的起点重跑这一段目标端用幂等写入覆盖。校验脚本的核心逻辑是这样import hashlib def checksum(rows, key_cols, val_cols): h hashlib.md5() for r in sorted(rows, keylambda x: x[key_cols[0]]): line |.join(str(r[c]) for c in key_cols val_cols) h.update(line.encode(utf-8)) return h.hexdigest() src fetch_mysql(SELECT id,user_id,amount FROM orders WHERE created_at BETWEEN %s AND %s, start, end) dst fetch_kafka(order_events, start, end) print(checksum(src, [id], [user_id, amount])) print(checksum(dst, [id], [user_id, amount]))key_cols用于排序保证两端顺序一致val_cols是要比对的业务字段。两个 checksum 相等说明这段数据一致不等就触发回放。注意 Kafka 侧拉数据要按 key 聚合后再比对否则分区顺序和数据库顺序对不上会误报。回放时把位点配置改成目标窗口起点启动中间件观察位点推进到窗口终点后停止任务。这一步的关键是目标端写入必须幂等否则回放会制造重复。我踩过的坑是回放窗口选得太大重跑时把线上正在消费的下游又灌了一遍后来改成按小时切片、逐片校验逐片回放影响面小很多。位点回放这套机制本质上是给同步链路留了一颗后悔药。没有它数据不一致时你只能全量重来几个 T 的表够你喝一壶。有了它问题定位到分钟级窗口回放几分钟就修完。做数据同步这行宁可前期多花两天把校验和回放搭好也别等出事时对着几千万行数据发呆。希望帮到你。本文还有配套的精品资源点击获取