联调最怕什么不是代码报错而是两边数据对不上。前端跟你说“渠道配置读的还是旧值”翻遍代码发现逻辑没问题最后定位到测试库里那张payment_channel表根本没有人把生产环境新加的几条记录同步过来。这种问题查代码永远查不出结果只能手动拼 SQL一行行导数据。次数多了你会忍不住想能不能写个工具一条命令把开发/测试环境的表数据快速对齐我花了一个下午用 Node.js 和 mysql2 写了一个轻量同步助手专门干这件事。它不像 DataX 那样需要 Java 环境和一堆配置文件也不像 Navicat 那样只能靠人工点鼠标。它就是一个可以放进项目仓库、可以挂在 CI 里跑的脚本指定源库、目标库、表清单就能把表数据对齐过去。这篇文章把设计思路、核心代码、踩过的坑都整理出来适合不满足于手动导表、希望把“数据对齐”这件事脚本化的后端和测试同学。1. 为什么我选择自己写同步脚本而不是用现成同步工具1.1 一次联调让我意识到“表数据对齐”会卡住整个流程当时的情况是三个服务联调A 服务依赖一张支付渠道配置表测试环境里的渠道还是旧的导致新联调的支付场景每次都在渠道校验那一步挂掉。后端说是数据问题前端说不是代码问题两边耗了一个小时。最后我用 Navicat 连上生产库把那张表导出再导入测试库问题才消失。这个场景其实很典型开发/测试环境不需要像生产环境那样追求高性能同步但需要“快速、可重复、无脑”地把数据对齐。所谓无脑就是不要每天在 Navicat 里点来点去不要对着几十张表手写 INSERT 语句更不要用 Excel 中转——导出来的时间格式可能就变了。1.2 现成方案各有各的尴尬我在动手前把能想到的方案都列了一遍结论是它们都不是为“开发/测试环境快速对齐单张表”这个场景设计的。方案优点缺点适合场景DataX功能全、性能强、支持多种数据源需要 Java 环境配置 json 较重大规模、跨异构数据源、离线同步Navicat 数据传输可视化、上手快重复劳动难以脚本化偶尔手工同步mysqldump原生、稳定粒度粗整库/整表操作容易覆盖本地数据整库备份、迁移自己写脚本轻量、可定制、可进 CI大数据量性能不如专用工具开发/测试环境、小到中量表数据DataX 是很好的工具但它太重了。为了把一张几百行的配置表从 A 库复制到 B 库我得写一个 DataX job 的 json里面还要配 reader、writer、通道数包一层命令行光学习成本就够我喝一壶。Navicat 能导但它是个 GUI 工具没法在自动化流程里调用而且不同环境的连接配置散落在不同人电脑上换台机器又要重新配。mysqldump 同样有个粒度问题——它适合把整个库导出来再恢复但开发环境往往不是空库你在上面改了字段、插了测试数据一个整库恢复就把本地东西全冲掉了。1.3 什么时候该自研、什么时候不该自研这不是在否定成熟工具。相反自研脚本一定要先想清楚边界。我总结下来的判断标准是这样单表数据量在百万行以内用 Node mysql2 的脚本完全能扛住。同步频率高比如每次开发联调前都要拉一次脚本可以做到“一条命令跑完全部表”。需要定制过滤条件比如只同步status 1的数据或者只同步最近一周修改过的数据脚本里加个 WHERE 条件就行。需要和现有项目仓库、CI/CD 流程结合脚本可以直接提交到 Git 仓库团队所有人的环境配置都能共享。反过来如果你面对的是几百 GB 的大表、需要跨机房跨云同步、需要断点续传和任务调度那还是用 DataX 或者 Flink CDC 这类专业工具。自研脚本的定位不是替代它们而是在“齐数据”这个高频、轻量的需求上少一些折腾。2. 核心设计流式读源表 批量写目标表2.1 从数据字典拿到表结构和可写列同步脚本要足够通用就不能把列名写死在代码里。最靠谱的做法是运行时从 MySQL 的information_schema里读取列信息。SELECT COLUMN_NAME, DATA_TYPE, EXTRA FROM information_schema.COLUMNS WHERE TABLE_SCHEMA ? AND TABLE_NAME ? ORDER BY ORDINAL_POSITION拿到列名之后要做两件事第一过滤掉auto_increment自增列如果源表有主键 ID同步时一般还是要带上后面细说第二过滤掉 MySQL 5.7 之后出现的虚拟生成列EXTRA里包含VIRTUAL或STORED。这些列不能显式写入否则 INSERT 会直接报错。有些场景下源表和目标表字段不一致比如目标表新增了一个源表里没有的字段或者源表有个字段对目标环境没意义。这时就不能盲目全列同步。我的做法是在配置里允许覆盖列清单默认情况下用information_schema自动选择特殊情况下手动指定await syncTable({ source: env.source, target: env.target, table: payment_channel, columns: [id, channel_code, channel_name, status] });2.2 为什么必须流式读取而不是 SELECT * 全部塞进内存最初版本我图省事直接用await connection.query(SELECT * FROM ...)几百万行数据一次性返回再组装成 INSERT 语句。结果进程内存冲到 2GB 以上在低配开发机上直接被 OOM kill。原因是 mysql2 默认把查询结果全部载入内存后才交给 Node.js数据量一大就扛不住。这是典型的“先灌满一水缸再抬过去”的做法。正确姿势应该是流式读取像自来水一样源库一边出数据这边一边攒批、一边写入目标端内存里始终只有一小批数据。mysql2 自带流式接口写法很简洁const query src.query(sql); const stream query.stream(); for await (const row of stream) { // 逐行处理 }这里有一个关键点src.query(sql)这一句不能加await加了就变成一次性返回所有结果流式就失效了。必须直接拿返回值里面的.stream()。2.3 目标端写入批量 INSERT 与事务控制流式读出来之后不能一行一行 INSERT那样太慢。正确做法是攒够一批再写。mysql2 支持把二维数组直接展开成批量 VALUESconst sql INSERT INTO \${table}\ (${colNames.map(c c ).join(, )}) VALUES ?; await tgt.query(sql, [batch]);batch是一个二维数组外层代表行内层代表列值。mysql2 会自动展开成VALUES (?, ?, ?), (?, ?, ?), ...非常方便。事务边界必须控制好。全量覆盖模式下流程是先DELETE FROM目标表再批量 INSERT。这两步必须放在同一个事务里否则 INSERT 中途报错目标表就变成半新半旧的状态比没同步之前更乱。await tgt.beginTransaction(); try { await tgt.query(DELETE FROM \${table}\); // 循环 flushBatch await tgt.commit(); } catch (err) { await tgt.rollback(); throw err; }2.4 一份可以直接跑的 sync-core.js 骨架把上面的思路串起来就是一个极简但完整的同步函数const mysql require(mysql2/promise); async function getColumns(conn, table) { const [rows] await conn.query( SELECT COLUMN_NAME, DATA_TYPE, EXTRA FROM information_schema.COLUMNS WHERE TABLE_SCHEMA ? AND TABLE_NAME ? ORDER BY ORDINAL_POSITION, [conn.config.database, table] ); return rows; } async function flushBatch(conn, table, colNames, batch) { const sql INSERT INTO \${table}\ (${colNames.map(c c ).join(, )}) VALUES ?; await conn.query(sql, [batch]); } async function syncTable({ source, target, table, where , columns null, mode replace, batchSize 500 }) { const src await mysql.createConnection({ ...source, supportBigNumbers: true, bigNumberStrings: true, dateStrings: true }); const tgt await mysql.createConnection({ ...target, supportBigNumbers: true, bigNumberStrings: true, dateStrings: true }); let colNames; let batch []; let readCount 0; try { const cols await getColumns(src, table); if (columns columns.length) { colNames columns; } else { colNames cols .filter(c !/auto_increment|VIRTUAL|STORED/i.test(c.EXTRA || )) .map(c c.COLUMN_NAME); } const sql SELECT ${colNames.map(c c ).join(, )} FROM \${table}\ ${where}; const query src.query(sql); const stream query.stream(); await tgt.beginTransaction(); try { if (mode replace) { await tgt.query(DELETE FROM \${table}\); } for await (const row of stream) { readCount; batch.push(colNames.map(c row[c])); if (batch.length batchSize) { await flushBatch(tgt, table, colNames, batch); batch []; } } if (batch.length) { await flushBatch(tgt, table, colNames, batch); } await tgt.commit(); } catch (err) { stream.destroy(); await tgt.rollback(); throw err; } return { readCount }; } finally { await src.end(); await tgt.end(); } } module.exports { syncTable };这段代码已经把核心链路完整呈现出来了读列信息、过滤不可写列、流式读取、批量写入、事务保护、连接回收。实际使用时你只需要在外面传source和target两个连接配置指定表名就能开始同步。3. 同步过程中最容易翻车的字段类型与连接配置3.1 BIGINT 和 DECIMAL 精度丢掉两行配置解决第一次跑真实业务表时我发现目标表里的雪花 ID 末尾几位全变成了 0。查了一圈才反应过来JavaScript 的Number能精确表示的整数只有 2 的 53 次方以内超过这个范围比如 19 位的雪花 ID就会被四舍五入精度悄悄丢掉。解决办法是给源库和目标库的连接都加上两个配置项supportBigNumbers: true, bigNumberStrings: truesupportBigNumbers让 mysql2 把超过安全范围的 BIGINT 当作字符串返回bigNumberStrings进一步保证 BIGINT 类型一律以字符串返回。字符串写入 MySQL 时MySQL 会重新解析成准确的 BIGINT数据就保住了。DECIMAL 字段默认返回字符串这其实是个好消息因为浮点数在 JS 里的精度问题尤其严重如果把它读成 number再序列化回去很可能出现 0.1 0.2 不等于 0.3 这类问题。所以除非你明确知道自己在做什么否则不要轻易开decimalNumbers: true。3.2 TIMESTAMP 时区偏移源头规避比事后修复省事另一个让我头疼的问题是同步完以后目标表里的时间字段差了好几个小时。查原因集中在 mysql2 的日期处理机制上。mysql2 默认会把 MySQL 返回的TIMESTAMP/DATETIME转成 JavaScript 的Date对象它的时区转换基于连接配置里的timezone参数默认是本地时区。如果源库和目标库的时区设置不一致或者运行脚本的机器时区与数据库时区不同数据在“字符串 - Date 对象 - 字符串”的转换过程中就可能产生偏移。最省事的规避方式是在连接配置里写死dateStrings: true这样 mysql2 会把时间字段直接以原始字符串形式返回不做任何 JS Date 转换同步脚本只做搬运工时间是什么样就搬什么样。代价是你拿到手的不再是 Date 对象但对我们这种纯搬运场景来说字符串反而是最安全的中间格式。3.3 自增列、虚拟列、NULL 与 JSON 字段的处理自增列要分情况看。如果源表有主键 ID同步时我一般会显式带上 ID因为目标表可能会有其他业务数据引用这些 IDID 变了关联关系就全断了。全量 DELETE 后显式插入原 IDMySQL 会自动把AUTO_INCREMENT推进到max(id) 1不会出现主键冲突。虚拟生成列是 MySQL 5.7 之后引入的EXTRA字段会标记为VIRTUAL GENERATED或STORED GENERATED这类列不能出现在 INSERT 语句里否则直接报错。代码里我已经用正则过滤掉了。NULL 和 JSON 字段一般情况下不用特殊处理。mysql2 对 JSON 列会自动解析成 JS 对象插入时再序列化回去日常使用没问题。但要注意一个边界情况如果源表字段允许 NULL目标表字段却设置成了 NOT NULL 且没有默认值同步就会在写入时失败。遇到这种表先检查一下两边的表结构是否一致或者把要同步的列用手动columns参数明确指定不要全列一把梭。3.4 max_allowed_packet 把大批量干断连批量大小不是越大越好。我把batchSize从 500 调到 5000 后同步几分钟就报一次错错误信息是连接被断开MySQL 服务端返回 packet 太大的提示。原因很简单MySQL 服务端有一个max_allowed_packet参数控制单个网络包的最大大小。如果一条 INSERT 语句拼接了 5000 行数据整个 SQL 的字节数可能远超这个限制服务端就直接断开连接。官方默认配置在一些低版本 MySQL 里是 4MB不是你想发多少就能发多少。遇到这个问题最稳妥的调整是调小batchSize比如压到 500 或者 200包体大小自然降下来。调大服务端max_allowed_packet不是不行但那是 DBA 要评估的操作开发环境犯不着。4. 从复制变成工作流增量同步和多环境配置4.1 什么场景用全量覆盖什么场景用增量同步不是只有一种模式。我日常用得最多的是replace全量覆盖模式适合配置表、字典表这类数据量小的表目标环境的数据本来就是乱的直接清掉重灌能恢复最干净的状态。但日志表、流水表不能这么干。比如一张用户登录日志表源库有 500 万行目标环境可能只需要最近几天的数据用于联调全量同步过去既慢又没必要。这种情况应该用增量同步只追新增和变更的数据。两个模式的分工很清楚小表、配置相关、要绝对对齐用全量覆盖大表、流水相关、只要追新数据用增量。4.2 updated_at 水位线方案增量同步最典型的实现是水位线。核心思路是找一张字段表里的时间字段通常是updated_at把这个字段的上次同步值记录下来下次同步时只取大于这个值的数据。配合INSERT ... ON DUPLICATE KEY UPDATE语法可以实现“存在就更新不存在就插入”的效果async function flushUpsert(conn, table, colNames, batch) { const updateCols colNames.filter(c c ! id); const placeholders batch .map(() (${colNames.map(() ?).join(,)})) .join(,); const sql INSERT INTO \${table}\ (${colNames.join(, )}) VALUES ${placeholders} ON DUPLICATE KEY UPDATE ${updateCols.map(c \${c}\ VALUES(\${c}\)).join(, )}; await conn.query(sql, batch.flat()); }这里有一点要提醒MySQL 8.0.20 开始VALUES()函数在ON DUPLICATE KEY UPDATE中有弃用警告官方推荐使用行别名语法。但如果批量语句用 mysql2 的展开方式行别名的兼容性反而容易出问题所以我在实际项目里暂时还是用的VALUES()因为 MySQL 5.7 和主流 8.0 版本都还能正常工作只是会产生一个 warning。水位线存在哪里也有学问。最简单的是存在本地文件里脚本跑完写一行 JSON{ user_login_log: 2025-01-20 12:00:00 }更正规的做法是在目标库里建一张sync_meta表把表名和上次水位线记在里面。这样不管脚本在哪台机器上跑水位线都能延续不会因为换环境就丢。增量同步配合 DELETE 旧数据的话逻辑会复杂很多开发/测试环境一般不用追求绝对一致能追新数据就够用了。4.3 一个 sync.config.js 管所有环境脚本要真正好用配置必须和代码分离。我把环境信息、表清单、同步模式都放到一个sync.config.js里module.exports { batchSize: 500, tables: [ { name: sys_config, mode: replace }, { name: payment_channel, mode: replace }, { name: user_login_log, mode: increment, watermark: updated_at } ], envs: { dev: { source: { host: 10.40.0.11, user: readonly, password: ***, database: prod_bak }, target: { host: 127.0.0.1, user: root, password: ***, database: dev } }, test: { source: { host: 10.40.0.11, user: readonly, password: ***, database: prod_bak }, target: { host: 10.40.0.32, user: root, password: ***, database: test } } } };主入口脚本用 Node 自带能力解析参数不需要引第三方库const args process.argv.slice(2); const getArg (name) { const i args.indexOf(--${name}); return i 0 ? args[i 1] : undefined; }; const env getArg(env) || dev; const tableNames (getArg(tables) || ).split(,).filter(Boolean);运行方式就变成了node sync.js --envdev --tablessys_config,payment_channel一个脚本所有环境的连接配置都在这一个文件里团队任何人拉到代码都能跑。源库建议只给SELECT权限的只读账号防止误操作写脏生产数据目标库用带写权限的账号。这个安全边界一定要守住。5. 实测数据与踩坑复盘5.1 百万行级别表同步实测最后说一组实测数据方便大家对这套脚本的性能有个体感。我在测试环境同步过一张 320 万行的订单流水表行平均大小约 800 字节源库和目标库在同一个局域网内MySQL 版本都是 8.0。配置batchSize 500开启事务跑完全量同步耗时约 43 秒进程最大内存保持在 180MB 左右。作为对比最初用SELECT *一次性加载的实现内存冲到 2GB 以上在机器上直接被系统杀掉。也就是说流式读取 批量写入的方案在百万行这个量级上完全够用。几百行的配置表就更不用说了基本秒开跑 CI 冒烟测试完全没压力。5.2 三个让我印象最深的坑第一个坑是内存爆炸。当时我以为问题出在数据量太大后来一步步排查才发现根源是await connection.query()把整个结果集都拉回来了压根没走流式接口。这也提醒了我一个原则只要是处理大结果集第一时间就要确认读取方式是不是流式的而不是先去调进程内存上限。第二个坑是外键约束。有一张主表和它的子表需要一起同步我先同步了主表主表数据被 DELETE 重灌后子表还在引用旧数据外键检查直接报错。后来在目标连接上加了一行await tgt.query(SET FOREIGN_KEY_CHECKS 0);同步结束再设回1。这里要注意FOREIGN_KEY_CHECKS是会话级变量只对当前连接生效所以必须在同一个连接上恢复否则当前连接关闭后配置就自动没了。第三个坑是时区偏移反复出现。最初我在目标环境排查数据时发现时间字段全部少了 8 小时一开始怀疑是代码 bug后来才发现是 mysql2 的 Date 对象转换逻辑导致的。添加dateStrings: true之后这个问题没有再出现过。5.3 还能往哪个方向扩展这个脚本最大的优势是可定制。你可以在它上面加差异报告对比源表和目标表的主键集合输出哪些行是新增的、哪些行是变化的、哪些行在目标环境被删掉了。也可以把它包成一个 npm script提交代码后自动把生产配置表同步到测试库。如果哪天数据量涨到千万级以上再引入 DataX 也不迟但在这之前这个轻量同步助手的性价比极高。我个人在实际使用中还有一个体会不要把连接配置写在代码里所有环境信息都应该走配置文件并且源库连接保持只读权限。我见过太多类似的脚本因为图省事把源库写成了可写账号某次手误把源库表清空的例子不是没有。同步工具本身是为了提效但安全底线不能因为效率而放松。