简介DBSyncer是一款开源数据同步中间件面向需要跨多种数据源同步的开发与运维人员解决MySQL、Oracle、SqlServer、PostgreSQL、ES、Kafka、File、SQL等异构数据源间的数据分发与转换问题。支持上传插件自定义同步逻辑并提供全量与增量同步的数据统计图、应用性能预警等可视化监控能力适合中高级后端开发者、DBA及数据平台建设者参考使用。包体共737个文件压缩包约2.07MB以472个Java源码文件为主体涵盖核心同步引擎与插件机制同时包含81个PNG图标或截图、59个CSS、42个HTML、36个JS等前端资源以及SQL脚本、XML配置文件、启动脚本与协议定义文件组成一套可运行、可扩展的完整项目模板。目前已有747人学习下载。通过该资源读者可深入理解多源同步中间件的模块划分与插件化设计思路掌握基于Web的控制台如何接入配置与监控模块也可直接修改插件扩展同步场景用于二次开发或技术选型验证。1. 为什么说开源数据同步中间件是异构数据库迁移的第一选择做数据同步中间件选型的人多半已经碰过这样的需求要把 MySQL 里的业务表倒进 Oracle 报表库把 SqlServer 老系统的数据搬到 PostgreSQL或者每天从文件服务器拉一批 CSV 同步到 Kafka。每个源库方言不同、类型体系不同靠写 JDBC 脚本硬搬能跑通一次跑不了一年——字段类型变了、增量范围错了、源库连接被断开哪一样都能让人凌晨起来救火。开源数据同步中间件解决的就是这个通用问题把“读哪个库、怎么变换、写到哪里”拆成可配置的插件化任务覆盖 MySQL、Oracle、SqlServer、PostgreSQL、File、Kafka 等常见数据源让一套配置代替一份定制脚本。适合正在搭数据中台、做库迁移或维护多套异构库的工程师也适合刚接手同步任务、想知道底层到底怎么跑的人。2. 拆解数据同步中间件的架构plugin 插件、channel 并发与限速2.1 reader / writer / transformer 三段式为什么同步中间件都长这样开源数据同步中间件虽然各家实现细节不同但主流的任务模型高度一致一个任务由 reader读端、writer写端和可选的 transformer转换器组成。reader 负责从源端拉数据writer 负责把数据写入目标端transformer 夹在中间做行级加工比如截断字段、脱敏、类型转换。理解了这个三段式再去看任何一款中间件的配置文件都不会懵。以常见的 DataX 系为例一个最小的任务配置长这样{ job: { content: [ { reader: { name: mysqlreader, parameter: { username: root, password: your_password, column: [id, user_name, phone], connection: [ { jdbcUrl: [jdbc:mysql://127.0.0.1:3306/biz_db], table: [t_user] } ], splitPk: id } }, writer: { name: oraclewriter, parameter: { username: sync_acc, password: your_password, column: [id, user_name, phone], preSql: [truncate table t_user_sync], connection: [ { jdbcUrl: jdbc:oracle:thin://127.0.0.1:1521/orcl, table: [t_user_sync] } ] } } } ], setting: { speed: { channel: 3 } } } }这段配置里reader 的 connection 是个数组意味着一个 reader 可以配多个连接串用于分库分表场景writer 的 preSql 在写入前执行常见做法是先清空目标表保证重跑不产生重复数据。setting.speed.channel 控制并发通道数通道数决定了同时有多少个线程在跑读写。不要把 column 和表字段的顺序搞错DataX 不会自动帮你对齐列名它按位置取值写入。2.2 splitPk 与 channel 数读端切分决定你的同步快慢channel 并发数设成 10任务不一定就能跑满 10 个线程前提是读端能切分出足够的任务分片。DataX 这类中间件切分任务依赖 reader 参数里的 splitPk指定一个主键或唯一键调度器把它按范围均分每个分片交给一个 channel 去读。如果表没有主键或者 splitPk 指向的列存在大量空值切分就会退化成全表扫描并发直接塌缩成 1~2 个通道。splitPk 的选型优先级一般是数字自增主键 数字业务键 字符串主键。字符串主键也能切但切分算法按字符串哈希走分片边界不好控制容易出现某个范围行数特别多、其他范围行数极少的情况导致部分 channel 空转。数值型主键最稳直接按 min 和 max 均分成 N 段每段范围行数基本均匀。channel 数也不是越大越好。每个 channel 会占用一个 JDBC 连接、一块读写缓冲目标库的连接数和单条事务大小才是瓶颈。我一般先看源库和目标库的规格源库是 4 核 8Gchannel 设 5~6源库是 16 核的物理机才考虑开到 10 以上。channel 设得过大源库连接数被打满同步速度反而下降而且数据库错误日志里会出现大量连接超时排查起来很费劲。2.3 为什么多数团队最终选 DataX 系而不是自己写同步自研同步工具最大的坑不是读写逻辑而是类型映射和异常恢复。MySQL 的 datetime 到 Oracle 的 DATE、Oracle 的 NUMBER 到 PostgreSQL 的 numeric看起来都是数字和日期实际 JDBC 驱动拿到的类型精度完全不同至少需要一张几十行的映射表才能兜住。DataX 这类开源工具已经把常见关系库、文件、消息队列的类型映射沉淀在插件里遇到没覆盖的类型报错信息也比自己写的清晰得多。另一个选择理由是增量同步和错误限速的成熟度。自研方案通常只处理“全量覆盖”这一种模式一旦业务方要求“每天只同步变更的数据”又要重新设计 where 条件的生成和上次同步位点的存储。开源中间件社区里这些场景都有现成方案不一定开箱即用但至少提供了可参考的插件实现。当然DataX 也不是没有短板它在实时性上不如 Flink CDC 这类流式计算框架配置是 JSON没有图形化界面。选型时先分清楚你要的是“定时批量同步”还是“秒级异构数据订阅”前者适合 DataX 系后者应该去看流式同步方案选错方向后面怎么调都是白费。3. 用 DataX 跑通 MySQL 到 Oracle 的最小同步任务job.json 与调参3.1 环境准备确认插件目录、驱动包和目标库连通性DataX 是解压即用的结构bin 目录下是启动脚本plugin 目录下按 reader 和 writer 分插件。拿到安装包后第一件事不是改配置而是确认 plugin/reader/mysqlreader 和 plugin/writer/oraclewriter 目录下有没有对应的 jar 包。常见翻车一个团队把旧版本的 DataX 放到服务器上直接用新任务配置跑结果找不到某个插件类其实只是安装包太老缺插件。启动命令是python3 bin/datax.py job/mysql_to_oracle.json建议先手动执行一遍确认能正常跑完再接入调度平台。DataX 启动时会把 JVM 参数打印到日志里很多新人会忽略这一段但里面能看到内存设置和类加载路径任务莫名卡住时查这里非常有价值。常见的做法是给足内存比如 4G 堆否则同步大表时频繁 Full GC速度肉眼可见地变慢。3.2 最小 job.json 长什么样从 MySQL 读、往 Oracle 写先拿一张小表跑通链路再上大表。下面这段配置是我日常起任务的最小编制加了必要的注释{ job: { content: [ { reader: { name: mysqlreader, parameter: { username: root, password: xxx, column: [id, user_name, phone, created_at], splitPk: id, where: created_at 2024-01-01 00:00:00, connection: [ { jdbcUrl: [jdbc:mysql://127.0.0.1:3306/biz_db?useSSLfalseuseUnicodetruecharacterEncodingutf8], table: [t_user] } ] } }, writer: { name: oraclewriter, parameter: { username: sync_acc, password: xxx, column: [id, user_name, phone, created_at], batchSize: 1024, preSql: [truncate table t_user_sync], postSql: [merge into t_user_stat using (select count(1) cnt from t_user_sync) s on (11) when matched then update set t_user_stat.cnt s.cnt], connection: [ { jdbcUrl: jdbc:oracle:thin://127.0.0.1:1521/orcl, table: [t_user_sync] } ] } } } ], setting: { speed: { channel: 4 }, errorLimit: { record: 0 } } } }reader 里的 where 参数建议写上即使全量同步也写一个恒真条件这样调增量时不用改结构只改 where 就行。writer 的 batchSize 控制每个事务写入多少行Oracle 批量写用 batch 机制值太小会频繁提交值太大会把回滚段撑爆一般从 512 到 2048 之间调。errorLimit.record 设成 0 表示有一条错误就失败退出适合数据异构迁移这种需要严格一致性的场景如果先跑历史数据、允许脏数据跳过可以放宽到百分比。3.3 三个必调参数byte、batchSize、channel 的选择逻辑task 跑得慢先别怀疑网络。多数情况是三个参数没配合好。第一个是 setting.speed.byte它是全局字节限速不是单通道限速。很多人把 byte 设成 1048576以为每个通道每秒传 1MB实际上整个任务被锁死在 1MB/s同步一张 20G 的表要跑五个多小时。如果不需要限速直接不写这个参数让通道数决定吞吐如果必须在业务高峰期限制对源库的压力再把 byte 和 channel 一起设。第二个是 writer 的 batchSize。它决定单事务写入行数。MySQL 写 Oracle 场景batchSize 建议 1024 起步观察目标库的 redo 日志切换频率和 undo 表空间增长再往上调整。Oracle 批量写如果遇到“ORA-30036: unable to extend segment by N in undo tablespace”就是 batchSize 太大或目标库 undo 太小调小 batchSize 或扩 undo 都可以。第三个是 channel 数它必须和 splitPk 的分片粒度匹配。一张没有 splitPk 的表channel 设 8 也没用读端只拆成一个任务8 个 channel 里 7 个空转。反过来一张 5 亿行的表 splitPk 正常channel 却只有 2读端依然会被拆成 2 个分片速度上不去。调到 channel 与分片数一致、且源库连接数没有报警就已经是当前环境下的上限。3.4 用 transformer 做列裁剪和脱敏而不是改上游视图有时源表里有手机号、身份证这类敏感字段同步到分析库之前要做脱敏。与其在上游数据库改视图你通常没有权限而且会影响别的消费方不如在同步任务里加 transformer。DataX 内置了几个常用转换器dx_col 可以按索引裁剪列dx_substr 可以截取字符串dx_seq 是自增序列。{ transformer: [ { name: dx_substr, parameter: { columnIndex: 4, beginIndex: 0, endIndex: 3 } }, { name: dx_col, parameter: { columnIndex: [1, 2, 4] } } ] }这个配置的意思是先把第 4 列从 0 数起截取前 3 个字符然后只保留第 1、2、4 列输出给 writer。要特别注意 columnIndex 是从 0 开始的很多人在这一步数错列结果脱敏了别的字段。transformer 是按行处理处理每一行都会产生一次函数调用行数特别大的时候会增加 CPU 消耗建议只在确有必要时使用不要在同步任务里做复杂的业务逻辑——那是数据清洗平台该干的事。4. 把同步场景扩到 SqlServer、PostgreSQL、File 与 Kafka一份配置一个数据源4.1 换库只改 readerSqlServer 与 PostgreSQL 的 jdbc 格式差异标题里提到的几个数据库在开源数据同步中间件里基本都是通过替换 reader 插件实现的。换数据源不复杂但 jdbcUrl 的写法是重灾区。SqlServer 的驱动串要带分号分隔的属性和 MySQL 的 连接风格完全不一样PostgreSQL 则要指定当前 schema否则同步报“relation does not exist”实际上表在默认 schema 下存在只是连接没切过去。{ reader: { name: sqlserverreader, parameter: { username: sa, password: xxx, column: [id, user_name], splitPk: id, connection: [ { jdbcUrl: [jdbc:sqlserver://127.0.0.1:1433;DatabaseNamebiz_db;encryptfalse], table: [t_user] } ] } } }SqlServer 的 jdbcUrl 里 encrypfalse 建议写上否则新版 JDBC 驱动默认开加密连本地开发库都可能会报 SSL 相关错误。PostgreSQL 的 reader 大同小异主要区别是驱动类不同url 写成 jdbc:postgresql://127.0.0.1:5432/biz_db?currentSchemapublic。如果表名带大写或特殊字符记得加双引号这一点和 MySQL 的习惯不同很容易被忽略。4.2 文件到 MySQLtxtfilereader 的路径、编码和空值三件事File 同步场景在平时不怎么起眼但一碰上批量导入历史数据它是最常用的入口。DataX 的 txtfilereader 读本地文件写入 MySQL 的 mysqlwriter。配置时有三个边界条件必须确认path 是目录还是文件、字段分隔符是什么、空值怎么表示。{ reader: { name: txtfilereader, parameter: { path: [/data/sync/2024/], encoding: UTF-8, column: [ {index: 0, type: long}, {index: 1, type: string}, {index: 2, type: string} ], fieldDelimiter: \t, nullFormat: \\N } } }path 写成目录时会读取该目录下所有文件且不递归子目录。如果文件是按日期分目录存放的要么在 path 里写多个目录要么把文件先统一放到一个目录。fieldDelimiter 用 Tab 时注意文件里是否混有空格否则会出现列错位。nullFormat 表示文件里什么样的内容当作 NULLMySQL 习惯用 \N这个值是从 MySQL 的导出格式继承来的正好覆盖了最常见的 CSV 导出场景。文件同步最常见的坑是编码不一致Windows 导出的 CSV 是 GBKLinux 上按 UTF-8 读出来中文全是乱码入库之后才发现。建议读端 encoding 先按文件实际编码填入库前统一在配置里转一次不要依赖数据库端的编码转换。4.3 Kafka 同步的两种常见落地路径开源数据同步中间件的 Kafka 接入方式和关系库不太一样。DataX 官方插件包里不带 kafkawriter社区实现的 Kafka 插件质量参差不齐。我一般把 Kafka 同步拆成两条路径看离线路径是先把 Kafka 消息落成文件再用文件同步进目标库准实时路径是直接用社区写好的 kafkawriter 把记录写入 Kafka topic。社区 kafkawriter 的配置通常是这种风格{ writer: { name: kafkawriter, parameter: { brokerList: 127.0.0.1:9092, topic: user_sync, batchSize: 1000, acks: 1, keySerializer: StringSerializer, valueSerializer: StringSerializer } } }batchSize 是 producer 端的批量发送阈值不是攒够 1000 条才发而是达到这个量级就触发发送。acks 设成 1 表示 leader 写入成功就返回吞吐高但极端情况可能丢数据同步场景我建议设成 all配合 Kafka 自身的副本机制保证不丢。还有一点要提醒社区插件对消息格式的约定各不相同有的要求 value 是 JSON 字符串有的要求是 Avro接入前先看插件的源码或 README不要盲目相信示例配置。如果同步的 Kafka 消息是从数据库 binlog 解析出来的字段名会随时间变化加列、删列这时不建议直接用 kafkawriter 按固定列写而是先落成 JSON 整体写入消费端再做 schema 兼容处理。这属于把同步和消费解耦的设计比在 writer 里硬扛结构变化靠谱得多。4.4 用 querySql 替代 table把 SQL 任务当同步源标题里提到“SQL 同步”最容易误解成“同步的时候执行 SQL 去更新目标库”。实际通常是指源端不直接给整表而是允许你用一条 SQL 查询作为数据来源。DataX 的 reader 参数里支持 querySql 和 table 二选一用了 querySql 就不能再配 table 和 column。{ reader: { name: mysqlreader, parameter: { username: root, password: xxx, querySql: [ select a.id, a.user_name, b.order_cnt from t_user a left join (select user_id, count(1) order_cnt from t_order group by user_id) b on a.id b.user_id where a.created_at 2024-01-01 ], connection: [ { jdbcUrl: [jdbc:mysql://127.0.0.1:3306/biz_db?useSSLfalse] } ] } } }用 querySql 最大的好处是可以做 join 和预聚合把清洗逻辑放在源库执行同步链路更短。但代价是失去了 splitPk 的并发切分能力整条查询结果只能由一个通道拉取大查询会非常慢。如果这个 SQL 查询的量级超过百万行建议不要用 querySql改成先把 join 结果落到临时表再用临时表做全量同步速度往往快一个数量级。另一个注意点是不同数据库的 SQL 方言差异Oracle 分页用 rownum 或 FETCH FIRSTSqlServer 用 OFFSET 和 TOPMySQL 用 LIMIT。同一个同步任务如果既要接 Oracle 又要接 SqlServerquerySql 必须按源库方言各自写一份。实际项目中更多是用 table 同步基础表用 querySql 同步统计结果表两者混用而不是一刀切。5. 数据同步中间件避坑指南5 条基于真实故障的排查记录5.1 同步条数对不上splitPk 指向了有空值的列现象任务日志显示成功但目标库的行数比源库少了几千行。复跑一次少的行数还不固定。原因splitPk 选了一个允许为空的字段比如 created_at。DataX 在切分范围时按该字段的 min/max 分段NULL 值的行无法被正确分到任何一段直接被跳过。表结构里这个字段大多数时候有值个别老数据是 NULL于是行数对不上。解决把 splitPk 换成主键或唯一非空字段。如果表确实没有主键先补一列自增 id 再同步或者改用全表扫描模式放弃并发切分保证数据完整优先。5.2 同步慢到翻车byte 限速被设成了单通道限速现象任务成功但速度只有几百行每秒和源库负载完全不匹配。原因setting.speed.byte 被设成 1048576也就是 1MB/s。这个参数是整个任务的总字节限速不是每个通道的限速。某个通道先跑够字节数整个任务就会等待所有通道共用这个配额。解决先确认限速意图。如果没有业务窗口限制直接删掉 byte 参数只保留 channel。如果一定要限速把 byte 设成 channel 数乘以单通道目标吞吐。比如 4 个 channel、每通道 5MB/sbyte 设 20971520。5.3 Oracle NUMBER 到 MySQL 变成科学计数法现象Oracle 里的 NUMBER(18) 字段同步到 MySQL 后变成了 1.23456789E10字符串和数值类型都乱了。原因JDBC 驱动读取 Oracle 的 NUMBER 时如果字段类型声明不明确会按 BigDecimal 或 float 处理。DataX 默认把它转成 Java 的 Double超过一定位数就输出科学计数法。解决在 reader 的 querySql 里把该字段先用 to_char 转成字符串或者用 CAST 指定为 VARCHAR 类型再同步。如果必须保留数值类型把目标表字段改成 DECIMAL(18,0)同时确认 reader 插件没有走默认类型推断的路径。这种类型映射问题靠调参数基本解决不了必须在 SQL 层先把类型定死。5.4 Kafka 同步分区不均写进去的消息挤在同一个分区现象Kafka topic 有 12 个分区同步任务跑完后某个分区消息数特别大其他分区几乎为空消费端吞吐上不去消息延迟高。原因kafkawriter 用表主键作为消息 key如果主键是自增 id哈希结果线性分布在分区数和 key 的哈希算法不匹配时大量 key 落在同一个分区。分区数如果和 key 的取模值互质关系不好热点就非常明显。解决在 writer 配置里把 key 改为业务字段组合或者给 key 加盐。如果不想在配置层做也可以在 SQL 查询阶段把主键拼接一个随机数后输出。消息在 Kafka 里的顺序性要求不高时对 key 做随机盐是最省事的方案。5.5 任务跑到一半连接断开MySQL 侧主动掐了连接现象大表同步跑了一两个小时日志里突然出现 Communications link failure任务直接失败。复跑多次总是在同一个时间点附近断。原因MySQL 的 wait_timeout 默认 8 小时但在内网环境很多 DBA 会把它调小到 2 小时。同步任务长时间空闲比如某个分片数据量特别小通道在等别的分片JDBC 连接超过阈值被服务端断开。用 Docker 起的 MySQL 更容易踩这个坑因为默认配置没调过。解决在 jdbcUrl 里加 socketTimeout 和 connectTimeout 参数值按最大单表同步时长估。同时让 DBA 把 wait_timeout 调大到 24 小时并确认 max_allowed_packet 不是限制瓶颈。任务侧重试次数也要配足DataX 的重试参数是针对单条插入失败的连接断开这类问题复跑时最好写个外层脚本做整体重试。6. 把同步接进生产校验、调度与一个值得养成的习惯6.1 一张核对表row count 加 checksum 怎么落同步任务跑完不等于数据对。我习惯每个任务后面额外起一个校验脚本先比 count再比 checksum。count 对不上直接失败count 对上了再抽样比对字段值。下面是一个简单的双库计数核对脚本#!/bin/bash # 先统计源库行数 src_cnt$(mysql -h127.0.0.1 -uroot -pxxx -N -e select count(1) from biz_db.t_user) # 再统计目标库行数 dst_cnt$(mysql -h127.0.0.1 -uroot -pxxx -N -e select count(1) from sync_db.t_user) if [ $src_cnt ! $dst_cnt ]; then echo count mismatch: src$src_cnt dst$dst_cnt exit 1 fi # 按主键分段比 checksum逐段输出差异数 mysql -h127.0.0.1 -uroot -pxxx -N -e select count(1) from biz_db.t_user s left join sync_db.t_user d on s.id d.id where d.id is null or s.user_name d.user_name or s.phone d.phone | read diff echo diff rows: $diffchecksum 全量比对大表代价很高所以分段比对比全表 join 更实用。上面脚本先找丢失的主键再比较关键业务字段diff 为 0 才算通过。这一步要放进调度平台而不是人肉执行否则漏一次就是事故。6.2 全量同步一次、增量五分钟调度上的常见搭配生产环境我一般把同步分成两级全量任务建表时跑一次增量任务定时跑。增量任务不需要 truncate 目标表把 reader 的 where 条件改成“created_at 上次同步位点”就行。同步位点可以由调度平台的上次运行时间决定也可以自己在目标库建一张 sync_position 表记录。用调度平台的时间作为位点有个好处不用额外建表坏处是调度平台时间不准时会出现重复或漏数所以我还是更推荐用源库的业务时间字段自行管理。增量任务的调度间隔五分钟是一个不容易出错的起点——太频繁会打满源库连接太稀疏会让下游报表数据滞后。Kafka 场景的准实时任务不经过这个定时器由 kafkawriter 持续写入。6.3 我养成的一个习惯每个任务一个 config 目录、一张监控看板早期我把所有同步任务的 job.json 堆在一个目录里改一个任务要 grep 半天。后来给每个任务建独立目录目录里放一个 README 写清楚源库、目标库、同步方式和业务负责人再配一个 run.sh 和 validate.sh。新增任务时复制一个模板目录改参数三分钟就能上线新同步链路。监控看板也不需要多复杂同步任务写一行的指标任务名、数据量、耗时、成功还是失败。Kafka 的 topic 分区情况单独看消费延迟高不高一眼就能看到。这个习惯让我在接手新库时特别省心看到目录就知道这个任务为什么存在、怎么改、失败了怎么验。同步中间件本身不解决管理问题但这些小规范能让配置变更和故障排查都走得更顺。希望帮到你。本文还有配套的精品资源点击获取
