做业务系统这么多年MySQL里面积攒的数据越来越多早晚会遇到一个绕不开的需求把数据从MySQL同步到别的地方去。可能是同步到数仓做离线分析可能是实时同步到消息队列给下游消费也可能是迁移到另一个数据库实例做读写分离。我习惯把这类需求统称为“MySQL数据出海”核心就是围绕MySQL构建一条可靠的数据流转管道。这篇文章不聊太虚的理论就聊聊我做过的几种主流的MySQL数据同步方案包括它们各自适合什么场景具体怎么落地以及实际操作中踩过的那些坑。1. “数据出海”到底在解什么题1.1 真实业务场景从报表系统说起的同步需求有段时间公司的数据分析团队天天找我抱怨说大屏上的GMV数据总是慢一个小时一到促销节点误差更大。当时的业务架构很简单订单库是MySQL数据分析平台用的是另一个存储中间靠一句定时任务每天凌晨三点跑批同步。结果就是运营同学白天看的实时数据和实际业务差着好几个小时促销期间干脆没法看。这种需求就是典型的数据出海。业务系统负责生产数据但数据的价值不只在生产端还要靠数据仓库、分析引擎、搜索引擎去放大。只要数据还锁死在MySQL里消费方拿到手的信息就永远是滞后的。这时候就需要一套机制把MySQL的数据增量地、准实时地搬运出去同时要保证数据不丢、不乱、延迟可控。1.2 同步方案要解决的三个核心问题我在评估同步方案的时候一般会盯住三个核心诉求这三个点也基本决定了技术选型的方向。第一是延迟。业务方对同步延迟的要求差异很大有的是分钟级报表有的是秒级实时风控有的干脆接受T1的离线批次。延迟需求决定了你是走binlog实时解析还是走定时任务批量抽取。第二是一致性。同步链路长了以后数据重复、乱序、丢数据的情况很常见。面试时候老问的“消息不会丢失”问题放在同步场景里也一样。你要考虑端到端如何保证数据不丢下游消费失败之后如何重试全量增量如何衔接。第三是对源库的影响。这个最容易忽略但恰恰最关键。如果同步方案对整个MySQL实例产生了明显的性能压力影响线上正常业务那这套方案不管数据多准都没意义。每种同步方式对源库的开销不同选型时一定得算清楚这笔账。2. 主流的四大同步路线逐个拆2.1 主从复制MySQL到MySQL的经典方案如果说数据出海的目的地还是MySQL比如做读写分离、容灾备份、分库分表后的汇总库那最合适的方案就是MySQL原生的主从复制。核心原理不复杂主库把变更写进binlog从库通过IO线程拉取binlog写入中继日志再由SQL线程回放最终实现数据一致。整个过程对业务方透明应用层不需要改任何代码。配置上无非是主库开启binlog、设置server-id从库执行CHANGE MASTER TO指向主库。深一点的内容比如半同步复制rpl_semi_sync_master_enabled、并行复制slave_parallel_workers、GTID复制等都是在原生复制框架上做增强和优化。这个方案最大的优点是简单可靠毕竟Oracle官方维护的东西稳定性在企业环境有保障。缺点也很明显它只解决MySQL到MySQL的问题。如果目的地是ClickHouse、Elasticsearch这条路就走不通了必须另想办法。2.2 binlog Canal实时增量同步的核心武器把MySQL的数据实时同步到Kafka、ES或者异构数据库业界最成熟的路线就是解析MySQL的binlog。而Canal就是基于binlog解析的明星工具。Canal的原理说白了是一个伪装成MySQL从库的Java进程。它向主库发送dump请求主库就当它是一个普通从库把binlog源源不断地推给它。Canal接到binlog后解析成结构化的事件再通过TCP或者MQ传输给客户端消费。这个方案的妙处在于完全剥离了业务代码。源库不需要做业务改造不需要双写一个进程挂在旁边就能把增量数据全量捞出来。而且binlog是MySQL底层机制只要事务提交了就会记所以说它能做到对线上业务侵入性最小的前提下拿到最完整的增量数据流。它的延迟可以做到毫秒到秒级完全取决于binlog的产生速度和消费端的处理能力。实测里单实例Canal的吞吐能力支撑每天上亿条变更的业务也没什么压力。2.3 ETL批量同步离线数仓的基本操作分析类场景里大量需求走的是离线同步。每天凌晨把昨天的数据从MySQL抽到数仓跑完ETL早上九点管理后台出报表。这类需求走实时方案是浪费走批量同步正合适。典型工具有DataX、Kettle、Sqoop这些。其中DataX是阿里开源的一个异构数据源离线同步工具我自己用得最多后面实操部分也拿它举例。批量同步的核心价值在于稳定和可控。任务跑在每天固定的时间窗口失败可以重跑数据可以全量刷也可以增量刷整个链路的排查思路非常清晰。但它解决不了实时问题而且如果业务量大到了像促销期间那种亿级表全量抽取对源库的压力会非常大要提前做好降级预案。2.4 业务双写万不得已的兜底方案还有一种方案是业务代码双写事务操作MySQL的同时把同一份数据写到另一个存储系统里。这听起来简单粗暴但实际维护成本很高。双写方案最明显的痛点是一致性无法保证。两边的写入不在同一个事务里一旦下游写失败或者一方成功一方失败数据就产生了不一致还得靠补偿任务来兜底。如果两边都写成功但中途发生网络抖动导致一方重复提交还得处理幂等。我的态度很明确双写只适合数据量不大、实时性要求高、目标端没有现成同步工具的极少数场景。但凡能走binlog方案就尽量不要让业务代码掺和同步这件事。3. 实操一Canal Kafka实时同步MySQL订单数据3.1 环境准备与前提条件先说前提MySQL必须开启binlog并且binlog格式要设成ROW。这一点极其重要我在后面踩坑部分会说为什么。先给出开启binlog的最小配置[mysqld] log-binmysql-bin binlog-formatROW server-id1配置完记得重启MySQL然后确认binlog已经生效SHOW VARIABLES LIKE log_bin; SHOW VARIABLES LIKE binlog_format;然后创建一个Canal专用的数据库账号。这个账号只需要授予复制权限就够了遵循最小权限原则CREATE USER canal% IDENTIFIED BY canal123; GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO canal%; FLUSH PRIVILEGES;接下来下载Canal部署包。到Canal的GitHub Release页面找canal.deployer的最新版解压后重点修改两个配置文件。3.2 配置Canal实例第一个文件是conf/canal.properties它控制整个Canal Server的属性。核心配置如下canal.port11111 canal.instance.master.address127.0.0.1:3306 canal.instance.dbUsernamecanal canal.instance.dbPasswordcanal123 canal.instance.connectionCharsetUTF-8 canal.instance.tsdb.enabletrue canal.instance.gtidonfalse canal.zkServerscanal.port11111是Canal Server对外提供服务的TCP端口客户端就是连这个端口拿数据的。canal.instance.master.address填MySQL的真实地址。注意这里虽然叫master但Canal还可以直接接从库的binlog只是为了减少对主库的压力我会把Canal指向从库地址。提示如果MySQL主库压力已经很大建议Canal直连从库因为dump binlog本身也是一份IO开销。从库回放完数据后binlog同样完整解析效果完全没有差别。第二个文件是conf/example/instance.properties它决定这个Canal实例监听哪些库表canal.instance.mysql.slaveId1234 canal.instance.master.positionmysql-bin.000001:4 canal.instance.filter.regexorder_db\\..*slaveId不能和MySQL已有的主从ID冲突。filter.regex的写法是库名\\.表名用正则匹配order_db\\..*就代表监听order_db库下面所有的表。注意反斜杠在properties文件里需要转义。3.3 Java客户端消费binlogCanal配置好之后启动服务sh bin/startup.sh服务日志里看到canal server is running now就算启动成功了。接下来写一个最简的Java客户端把binlog事件拉出来打印CanalConnector connector CanalConnectors.newSingleConnector( new InetSocketAddress(127.0.0.1, 11111), example, , ); connector.connect(); connector.subscribe(order_db\\..*); while (running) { Message message connector.getWithoutAck(100); long batchId message.getId(); try { for (CanalEntry.Entry entry : message.getEntries()) { if (entry.getEntryType() CanalEntry.EntryType.ROWDATA) { CanalEntry.RowChange rowChange CanalEntry.RowChange.parseFrom( entry.getStoreValue()); for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) { switch (rowChange.getEventType()) { case INSERT: System.out.println(insert: rowData.getAfterColumnsList()); break; case UPDATE: System.out.println(update: rowData.getBeforeColumnsList() - rowData.getAfterColumnsList()); break; case DELETE: System.out.println(delete: rowData.getBeforeColumnsList()); break; } } } } connector.ack(batchId); } catch (Throwable e) { connector.rollback(batchId); } }这个代码里的核心套路就是getWithoutAck拿到一批数据处理成功后ack处理失败就rollback让Canal重新投递。这样保证了数据不会因为客户端crash而丢失。实际生产中的消费端不会只打印而是把事件序列化成JSON后发送到Kafka由Kafka做消息缓冲下游再决定要不要清洗、转换、入库到数仓或ES。3.4 验证同步链路是否完整的技巧链路搭好之后得验证到底有没有漏数据。我有个自己常用的验证套路先在业务表里插入一条带唯一标识的测试数据然后立刻在消费端查看有没有对应的记录。更严谨的做法是做一个准实时的数据比对。因为Canal解析的是ROW格式的binlog所以拿到的是变更前后完整的行数据。把这个数据落到另一个MySQL表里然后定期执行一次COUNT(*)或CHECKSUM TABLE跟源库对账。对比有差异就说明同步链路出问题了。这里我强烈建议给消费端加一个延迟监控做法很简单源库每次变更时带上update_time字段消费端处理完以后计算当前时间 - update_time超过阈值就报警。这样延迟不再靠感觉而是有量化指标。4. 实操二DataX离线同步MySQL到数仓4.1 安装DataX并准备目的地再来看离线批量同步。这里用DataX演示。DataX的安装非常省事下载bin包解压即可不依赖额外的环境。目录结构大致是bin/datax.py执行入口job/存放任务JSONplugin/存放各种读写插件我的目标是每天凌晨把MySQL的订单表全量同步到HDFS再由Hive外部表分析。为了控制对源库的压力我会在MySQL侧配置一个从库作为数据源从库的数据本身就是全量的拿来抽取正合适。4.2 编写DataX任务配置DataX的同步任务核心是一个JSON配置。它的结构我心里有个公式记忆一个job下面有content对应的就是一条同步任务的reader和writer和setting控制速度和错误容忍度。下面这个配置就是把MySQL的订单表同步到HDFS{ job: { setting: { speed: { channel: 4 }, errorLimit: { record: 0, percentage: 0.02 } }, content: [ { reader: { name: mysqlreader, parameter: { username: datax, password: datax123, column: [id, order_no, user_id, amount, create_time], splitPk: id, connection: [ { table: [order_info], jdbcUrl: [jdbc:mysql://127.0.0.1:3306/order_db?useSSLfalse] } ] } }, writer: { name: hdfswriter, parameter: { defaultFS: hdfs://hadoop-namenode:8020, fileType: orc, path: /warehouse/order_db.db/order_info, fileName: order_info, column: [ {name: id, type: bigint}, {name: order_no, type: string}, {name: user_id, type: bigint}, {name: amount, type: double}, {name: create_time, type: string} ] } } } ] } }这份JSON里几个参数值得说道说道。splitPk这个参数我特别看中。意思是告诉DataX用哪个字段去做切分实现并行读取。比如传idDataX会把WHERE id ? AND id ?自动拆成多个区间分到不同的channel并发去查。没有splitPkDataX就单通道读取速度差很多。channel是并发数。我一般不建议盲目调大因为通道越多对源库的连接压力越大。线上实例如果是4核8G建议channel先给4到6观察源库的CPU和连接数没有异常再往上加。errorLimit控制错误容忍度。生产环境我会设成record: 0也就是任何一条记录失败整个任务失败宁可用重跑来保证准确也不要带病偷偷跑完。4.3 执行任务与控制并发配置写好后执行python bin/datax.py job/mysql_to_hdfs.json执行过程中DataX会实时打印同步进度包括每秒读取的记录数、同步总条数、耗时等。跑完之后对比一下源表和目标表的总行数是否一致。关于并发和限流我再补充一点实操心得。DataX的限流有两个手段一是调speed.channel二是调reader里的queryTimeout和连接参数。实际项目中我更推荐用channel控制并发因为它是全局级别的限流效果直观。如果同步任务和线上业务高峰期撞车了最好错峰或者在任务里临时把channel调到2扛过去再说。增量同步的写法也不难在reader的SQL里用WHERE create_time 昨天的日期当作过滤条件每天任务跑完记录一下当前水位下一次同步从水位开始。这样比全量同步的效率高很多对源库的压力也小。5. 踩坑实录与排查技巧速查5.1 binlog环节的几个经典问题先说binlog格式问题。如果你的MySQL binlog格式是STATEMENT而不是ROWCanal解析出来的就不是完整的行数据而是SQL语句本身。这让UPDATE和DELETE之后的“变更前数据”无从谈起整个同步链路的准确性直接崩掉。解决办法就是把binlog格式改成ROW同时确认它不会随着重启被MySQL的默认配置覆盖。再说权限不够的问题。Canal账号如果没有REPLICATION SLAVE权限连接时就会报Access denied for user canal。这个报错很常见排查步骤就两步先用SHOW GRANTS FOR canal%确认权限再用Canal自带的客户端工具从MySQL手动dump一次binlog验证。还有一个让人头疼的问题Canal只能拿到配置之前的binlog。如果Canal第一次启动时指定master.position对应的位点文件已经被MySQL清理掉了它会自动跳到当前最新的位点中间的那段数据就丢了。我的建议是要么配置任务前手动查询SHOW MASTER STATUS记录当前位点并确保binlog保留时间够长要么干脆接受从当前时刻开始同步的事实然后先把全量数据导到目的地增量再从配置好那一刻开始追。5.2 数据一致性问题的实战解法同步链路里最怕的数据不一致有几种表现目标端少数据、目标端数据重复、目标端字段对应错位。少数据一般出在消费端rollback用太多。如果ack一直不调用Canal会把同一批数据无限再次投递但你自己的代码可能把一部分数据成功写库了一部分没写再重投一遍就重复了。我的做法是消费端做成幂等写目标表加唯一索引写入用INSERT ... ON DUPLICATE KEY UPDATE这样即使重复投递结果也是收敛的。字段错位则是坑在column配置。DataX里reader的column顺序要和writer对齐如果中途业务改了表结构比如新增字段但任务配置没跟着改就会写入失败。这里我建议把目标端同步任务纳入遇到表结构变更时的必改清单实际上线前拿一条真实记录做个端到端验证最稳妥。5.3 性能瓶颈源库记得降级目标端记得合并最后说性能。实时同步链路中Canal本身对源库的额外压力很小但如果你是直连主库并且正好赶上大事务比如一条UPDATE影响几十万行binlog体量会瞬间暴涨从主库dump这些binlog的流量非常可观。线上让Canal连从库是最稳妥的止损方式。离线同步的性能瓶颈往往在目标端。DataX用MySQL作为writer时如果一次性insert几千行在目标库没有开启批量提交的情况下一行一次网络往返慢得离谱。DataX的mysqlwriter默认就开启了批量提交preSql和批量配置但我自己写JDBC批量写入的时候仍然建议显式使用rewriteBatchedStatementstrue这个参数能把多条insert编译成一条多VALUES的SQL实测性能提升非常明显。还有一点目标端如果是HDFS小文件问题要注意。DataX并发通道多的时候容易产生几百个几KB的小文件对Hive查询很不友好。解决方案是控制channel数量或者同步完成后跑一次任务合并小文件。6. 方案选型总结与个人建议最后聊聊怎么选。我自己做选型时会画一张简单的决策表虽然这里不用图表但逻辑很容易说清楚。目的地是MySQL首选原生的主从复制这没什么好纠结的。目的地是异构存储且对实时性有要求选binlog Canal这是目前最主流的实时数据出海路线。目的地是数仓对实时性不敏感选DataX之类的批量ETL工具。数据量小、临时性需求、实时要求高可以临时用双写兜底但心里要清楚它带来的数据一致性隐患用完后最好尽快切换到正式方案。我的个人体会是不要试图用一个同步工具包打天下。我之前接过一个项目既要实时报表又要离线分析硬塞了一套组件去扛所有场景结果实时链路因为离线任务全量抽取占了源库IO延迟飚到了几十秒。后来改成两条独立链路——实时走Canal离线走DataX——两边互不干扰各自调度问题很快解决。最后分享一个我在运维阶段养成的习惯给每个同步链路建立单独的监控大盘指标就三个——延迟时间、消费位点积压量、错误记录数。延迟超出阈值报警积压量持续上涨说明下游消费能力不够错误量突增说明可能有脏数据。平时多花点心思在监控上比出了事故再排查要省心得多。同步方案没有银弹但只要你把延迟需求、一致性要求和源库压力这三件事提前想清楚选择就不会跑偏。希望这篇总结能帮到正在折腾MySQL数据同步的人。