DBSyncer 数据同步中间件实战：MySQL 到 Elasticsearch 全量增量监听配置与避坑指南

简介DBSyncer简称dbs是一款开源的数据同步中间件面向需要跨库、跨平台数据流转的开发者与运维人员解决MySQL、Oracle、SqlServer、PostgreSQL、Elasticsearch、Kafka、File、SQL等多种数据源之间的全量与增量同步问题。资源包共737个文件以472个Java源码为核心辅以html、css、js等前端页面资源以及xml、sql、json、sh、bat等配置与启动脚本整体约2.07MB结构完整便于二次开发与本地部署。项目支持上传插件自定义同步转换业务并提供全量、增量数据统计图与应用性能预警等监控能力适合研究数据同步架构、插件扩展机制与监控告警实现的读者参考。目前已有747人学习下载可从中获取同步场景设计思路、插件化改造方法与监控模块实现细节对搭建企业级数据同步链路具有实用价值。1. 数据同步中间件选型为什么我最终把 DBSyncer 留在了生产环境上周有个做数据中台的朋友找我说他们要把 MySQL 的订单表实时同步到 Elasticsearch 做检索中间还得过一层 Kafka 做缓冲问我有没有轻量点的方案。我第一反应是 Canal 加一堆自研消费者但运维成本太高DataX 又只能跑批做不到准实时。后来翻到自己两年前拆过的 DBSyncer重新拉起来跑了一遍发现它把 MySQL、Oracle、SqlServer、PostgreSQL、ES、Kafka、File、SQL 这些同步场景都收在一个控制台里还带全量增量统计图和性能预警确实省事。这篇就把我拆包、部署、配驱动、调同步任务的完整过程写下来顺带把几个能让人卡半天的坑说清楚。适合手里有异构数据源、又不想上重型商业工具的同学照着复现。2. DBSyncer 的架构拆解驱动、连接器与同步模型怎么对上号2.1 从启动脚本看它到底装了什么拿到包之后别急着双击先看根目录那几个文件。startup.bat和startup.sh是启动入口version.cmd用来打印版本build.cmd是源码构建用的。真正干活的是lib下的 jar 和plugins目录里的驱动包。DBSyncer 本身不捆绑任何数据库驱动MySQL、Oracle、SqlServer、PostgreSQL 的 JDBC 驱动都得自己丢进plugins对应子目录这一点和很多“开箱即用”的宣传不一样但反过来也避免了驱动版本冲突。我一般会先跑一遍version.cmd确认版本再检查plugins下有没有mysql、oracle、sqlserver、postgresql四个文件夹。没有就手动建把对应驱动 jar 放进去。Elasticsearch 和 Kafka 的连接器在lib里已经带了不用额外加。前端资源里能看到bootstrap.min.css、font-awesome.min.css、_all.css这些说明控制台是典型的 Bootstrap 风格浏览器兼容性不用太担心。2.2 同步模型全量、增量、监听三件事分开做DBSyncer 把同步拆成三种驱动类型全量、增量和监听。全量就是一次性把源表数据搬到目标表适合初始化增量是基于时间戳或自增 ID 拉取变化数据适合定时补数监听则是通过数据库日志MySQL 的 binlog、Oracle 的 LogMiner 等捕获实时变更。很多人一上来就想配监听结果源库没开日志或者权限不够直接卡住。我的建议是先用全量把历史数据灌进去再切增量跑一段时间验证数据一致性最后才上监听做实时。这样每一步都有回退余地。配置的时候注意全量驱动和增量驱动可以绑同一个源表和目标表但监听驱动需要单独建因为它的位点管理和前两者不共享。2.3 连接器配置JDBC URL 和账号权限的硬性要求每个数据源都要在控制台里建连接。MySQL 的 URL 我一般写成jdbc:mysql://host:3306/db?useSSLfalseserverTimezoneAsia/ShanghaicharacterEncodingutf8注意serverTimezone不写会报时区错误。Oracle 用jdbc:oracle:thin:host:1521/ORCLSqlServer 用jdbc:sqlserver://host:1433;DatabaseNamedbPostgreSQL 用jdbc:postgresql://host:5432/db。账号权限方面源库账号至少要有SELECT和REPLICATION SLAVEMySQL 监听场景目标库账号要有INSERT、UPDATE、DELETE和建表权限。Oracle 监听还需要LOGMINING权限。这些在官方文档里写得比较散我踩过一次坑MySQL 账号只给了SELECT全量能跑一开监听就报权限不足查了半天才定位到。-- MySQL 源库授权示例 GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO dbsync%; FLUSH PRIVILEGES;这段授权里REPLICATION SLAVE是监听 binlog 必须的REPLICATION CLIENT用来查位点。如果只跑全量和增量这两个可以不给但建议一次性配好免得后面切监听再改。3. 从零跑通一条 MySQL 到 Elasticsearch 的同步链路3.1 环境准备与启动先确认 JDK 版本DBSyncer 要求 JDK 8 或 1117 以上会有模块化报错。我一般用 JDK 11。下载包解压后Linux 下执行chmod x startup.sh然后./startup.sh。Windows 直接双击startup.bat。启动成功后控制台默认监听 18686 端口浏览器打开http://localhost:18686就能看到登录页默认账号密码在application.yml里一般是admin/admin。如果启动报端口占用改application.yml里的server.port。如果报驱动加载失败检查plugins目录结构必须是plugins/mysql/mysql-connector-java-8.0.xx.jar这种层级不能把所有 jar 平铺在plugins根目录。3.2 建源连接和目标连接登录后进“连接管理”先建 MySQL 源连接。填名称、类型选 MySQL、填 URL、用户名、密码点测试连接。通了之后建 Elasticsearch 目标连接类型选 Elasticsearch填集群地址比如http://host:9200。如果 ES 开了认证URL 里带用户名密码或者单独填。这里有个细节ES 连接器默认走 HTTP如果集群是 HTTPS需要在 URL 里写https://并在高级参数里配证书信任。我一般在内网环境直接用 HTTP省事。3.3 配置同步驱动全量灌数据进“驱动管理”新建驱动类型选全量。源连接选 MySQL目标连接选 ES。然后配表映射源表选order目标索引填order_index。字段映射可以自动匹配也可以手动改。注意 ES 的_id默认用源表主键如果源表没有主键得手动指定一个唯一字段否则会重复插入。{ sourceTable: order, targetIndex: order_index, idField: order_id, fieldMappings: [ {source: order_id, target: order_id, type: long}, {source: user_name, target: user_name, type: keyword}, {source: amount, target: amount, type: double}, {source: create_time, target: create_time, type: date} ] }这段映射里idField决定 ES 文档的_id不配会用默认生成值导致重复同步时数据翻倍。type字段控制 ES 的 mapping 类型keyword和text别搞混前者用于精确匹配后者用于全文检索。配完点“启动”全量任务开始跑。控制台会显示进度条和已同步条数。如果卡住不动看日志里有没有BulkRequest超时一般是 ES 的refresh_interval太短或者批量太大把批量从 1000 调到 500 试试。3.4 切增量时间戳字段和定时策略全量跑完后新建一个增量驱动源表还是order但增量字段选update_time。DBSyncer 会记录上次同步的最大时间戳下次从那个点往后拉。定时策略我一般设 1 分钟一次太频繁会给源库压力太慢又失去准实时意义。增量同步的坑在于时间戳精度。MySQL 的datetime默认秒级如果同一秒内有多次更新增量可能漏数据。解决办法是把字段改成datetime(3)毫秒级或者在增量 SQL 里加而不是配合去重逻辑。我一般直接改表结构上毫秒精度省心。3.5 监听模式binlog 位点与断点续传监听驱动依赖 MySQL 的 binlog。先确认my.cnf里log-binmysql-bin、binlog_formatROW、server_id1都配了。然后建监听驱动选源表和目标表启动后 DBSyncer 会从当前位点开始读。断点续传是自动的位点存在内置的 H2 数据库里。但如果服务重启后位点丢了检查data目录权限H2 文件写不进去会导致位点重置进而重复同步。我一般把data目录挂到独立磁盘避免和系统盘抢 IO。4. 避坑与排查驱动、权限、字段类型这三关最容易翻车4.1 驱动版本不匹配导致连接测试失败现象建 MySQL 连接时点测试报No suitable driver found或Communications link failure。原因plugins/mysql下的驱动 jar 版本和数据库版本不匹配比如 MySQL 8.0 用了 5.1 的驱动。解决MySQL 8.0 用mysql-connector-java-8.0.28以上5.7 用5.1.49。Oracle 用ojdbc8SqlServer 用mssql-jdbc-9.4.1.jre11。放进去后重启服务。4.2 源库账号权限不足全量能跑监听报错现象全量同步正常一开监听就报Access denied; you need REPLICATION SLAVE privilege。原因账号只给了SELECT。解决按 2.3 的授权语句补REPLICATION SLAVE和REPLICATION CLIENT然后FLUSH PRIVILEGES。Oracle 的话需要GRANT LOGMINING TO dbsync;。4.3 字段类型映射错误导致 ES 写入失败现象同步到 ES 时报mapper_parsing_exception比如把字符串写进了long字段。原因自动映射时源表varchar被识别成text但目标索引里已经建了keyword或long。解决在驱动配置里手动改字段类型或者在 ES 里先删索引重建。我一般同步前先让 DBSyncer 自动建索引避免类型冲突。4.4 增量同步漏数据时间戳精度和时区问题现象增量跑完后对账发现少了几条。原因update_time是秒级同一秒内多次更新只捕获到最后一次或者源库时区和 DBSyncer 时区不一致。解决字段改毫秒精度JDBC URL 里加serverTimezoneAsia/ShanghaiDBSyncer 启动参数加-Duser.timezoneAsia/Shanghai。4.5 Kafka 目标端消息延迟高现象同步到 Kafka 后消费端延迟越来越大。原因Kafka 生产者batch.size和linger.ms配置保守或者分区数太少。解决在 DBSyncer 的 Kafka 连接高级参数里把batch.size调到 16384linger.ms调到 10同时把 topic 分区数加到 6 以上。如果还慢看 Kafka 集群磁盘 IOkafka 读写最大值与硬件关系这个热词说的就是磁盘瓶颈。5. 进阶技巧用 SQL 驱动做跨库转换和自定义插件5.1 SQL 驱动不写代码做字段清洗DBSyncer 的 SQL 驱动允许在同步过程中执行自定义 SQL。比如源表order的status是数字目标 ES 要存中文可以在驱动里配转换 SQLSELECT order_id, user_name, CASE status WHEN 1 THEN 待支付 WHEN 2 THEN 已支付 WHEN 3 THEN 已发货 ELSE 未知 END AS status_text, amount, create_time FROM order WHERE update_time :lastUpdateTime:lastUpdateTime是 DBSyncer 内置的增量参数会自动替换成上次同步的时间戳。这样不用改源表结构也不用写 Java 插件直接在 SQL 里做映射。注意 SQL 驱动只支持标准 SQLOracle 的decode和 SqlServer 的iif也能用但别用存储过程DBSyncer 不解析。5.2 自定义插件上传 jar 扩展转换逻辑如果 SQL 搞不定比如要调外部 API 做数据脱敏就得写插件。DBSyncer 的插件接口是com.dbsyncer.plugin.Plugin实现convert方法打成 jar 丢进plugins/custom目录重启后在驱动配置里选自定义插件。我写过一个手机号脱敏的插件核心就十几行public class PhoneMaskPlugin implements Plugin { Override public MapString, Object convert(MapString, Object source) { if (source.containsKey(phone)) { String phone (String) source.get(phone); if (phone ! null phone.length() 11) { source.put(phone, phone.substring(0, 3) **** phone.substring(7)); } } return source; } }编译时把 DBSyncer 的dbsyncer-plugin-api.jar加进 classpath打包后放对目录。注意插件里别做耗时操作否则会拖慢整个同步链路。我一般把 API 调用改成异步或者提前把映射关系加载到内存。5.3 监控与预警全量增量统计图怎么看控制台的“监控”页有全量和增量的统计图横轴是时间纵轴是同步条数。如果增量曲线突然掉到 0说明源库没新数据或者监听断了。性能预警可以配阈值比如同步延迟超过 60 秒发邮件。我一般把预警接到企业微信机器人出问题第一时间知道。验证数据一致性的话我习惯在目标库跑 count 和 checksum和源库对比。ES 的话用_countAPI 查文档数再用_search抽样比对字段值。别全量比对太慢抽样就行。从那以后我每次配同步任务都强制先跑全量、再切增量、最后上监听三步走完才敢放到生产。这套流程帮我省了至少三次半夜爬起来修数据的时间。希望帮到你。本文还有配套的精品资源点击获取