这两年做搜索和数据分析类的项目几乎绕不开一个老生常谈的问题业务数据都躺在 MySQL 里可一旦查询变复杂——多条件筛选、全文检索、聚合统计——MySQL 的执行计划就开始力不从心。于是很自然想到把数据搬到 Elasticsearch让 ES 来扛查询压力。但问题紧接着就来了怎么把 MySQL 里的数据干干净净、低延迟地同步到 ES我试过定时任务全量刷也试过各种开源同步工具最后稳定跑下来的方案是用 Canal 订阅 MySQL 的 binlog把增量变更事件解析出来再写入 Elasticsearch。这篇就是把这条链路从零搭起来、踩过坑、又调优过的完整记录适合正在做搜索服务、数据报表或异步查询拆分的后端开发也适合刚开始接触 Canal 的运维和架构同学。1. 为什么 MySQL 到 ES 的实时同步能解决搜索慢这个根本问题先搞清楚一件事我们做同步不是为了把数据复制一份出来好看而是要解决一个具体痛点——MySQL 不擅长复杂查询ES 擅长。但数据在两个系统里如何保持一致就成了架构设计里绕不开的一环。1.1 一张表一条 SQL能在 ES 里对应出什么MySQL 里一个典型场景一张订单表关联了用户表、商品表、店铺表前端要做一个订单搜索页支持按订单号、用户昵称、商品名称、日期范围、金额范围过滤还能按店铺聚合统计。这种需求如果直接落在 MySQL 上往往要 join 三四张表再加LIKE %xxx%、范围查询和GROUP BY索引很难同时生效数据量一旦到百万级响应时间就很难看。而 ES 的做法是把多表数据在写入时做一次宽表化——把用户昵称、商品名称、店铺名这些字段都冗余进一个嵌套文档里查询的时候就变成了单索引的单次检索完全绕开了 join 的成本。同步的本质就是保证 MySQL 业务库里更新的数据能及时变成 ES 索引里一份适合检索的结构化文档。1.2 为什么选 Canal 做数据管道而不是定时任务很多项目最早都是定时任务全量刷。比如每五分钟把整张表读一遍 upsert 进 ES。但全量刷的问题非常明显数据量大了以后五分钟刷一次根本扛不住而且每次全量读取对数据库和网络都是一次压力测试更麻烦的是删除操作很难被捕获——除非用软删除字段否则 ES 里那条文档永远删不掉。Canal 的思路完全不同。它直接订阅 MySQL 的 binlog每次增删改都会实时产生一条变更事件Canal 解析后推给下游。这样做有几个优势实时性高从数据库变更到 ES 更新通常能控制在一秒以内。数据库压力极小只是模拟一个从库去拉 binlog不额外做查询。删除操作也能被捕获因为 binlog 里记录的是DELETE的真实主键。所以从选型角度看凡是要求联动更新的场景比如库存变化后商品搜索页要立刻可见、订单状态变更后后台列表要马上刷新Canal 都是比定时任务更合理的底座。当然它也有自己的复杂度——部署 Canal、维护消费逻辑、处理异常、保证幂等——这些正是后面要展开的部分。2. 环境准备binlog、账号权限和 ES 索引规划很多人第一步就被卡住不是不会写代码而是 MySQL 那边没打开 binlog或者开了 binlog 但格式不对。这里说的格式不是小事情它直接决定 Canal 能不能拿到每一行具体变成了什么。2.1 打开 binlog 并切换到 row 格式MySQL 的 binlog 有三种格式STATEMENT、ROW、MIXED。Canal 要的是ROW格式因为它需要在 binlog 里直接看到插入或修改后的整行数据。如果用的是STATEMENT格式日志记录的是执行了什么 SQLCanal 无法知道数据到底变成了什么样。在 MySQL 的配置文件如my.cnf里需要确认以下配置server_id100 log_binmysql-bin binlog_formatROW binlog_row_imageFULL expire_logs_days7 max_binlog_size256M这里的几个参数分别对应什么我也梳理成一张表方便对照参数作用需要注意的点server_id在复制拓扑中标识当前实例建议独立不会重复log_bin开启 binlog 日志修改后需要重启 MySQLbinlog_formatROW记录行级变更Canal 的硬性要求binlog_row_imageFULL记录变更前后的完整镜像否则可能拿不到旧值expire_logs_days控制 binlog 保留时间太短会导致同步断点追不回来max_binlog_size单个 binlog 文件大小太大不方便轮转太小会增加 IO提示不是所有环境都能随意重启 MySQL。如果用的是云厂商的托管数据库一般可以在参数组里在线修改binlog_format并手动触发轮转但还是要先确认是否允许模拟从库拉取 binlog。另外还要给 Canal 准备一个专用的 MySQL 账号权限至少包括SELECT和REPLICATION SLAVE、REPLICATION CLIENT。这样 Canal 才能以从库的身份去拉日志CREATE USER canal% IDENTIFIED BY your_password; GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO canal%; FLUSH PRIVILEGES;2.2 Canal Server 部署的关键参数Canal 本身的部署分为 Server 端和 Client 端。Server 端负责连接 MySQL、解析 binlog、把变更包装成统一的事件模型Client 端负责消费这些事件并写入 ES。部署方式有两种常见选择直接用canal-deployer跑一个独立进程。在 Kubernetes 里用容器方式部署方便扩容和与业务系统共存。我的经验是初期单机部署一个 Canal Server 就够用因为瓶颈通常不在 Canal 本身而在下游 ES 的写入能力。真正做生产化的时候才需要考虑多实例、内嵌版或者接入消息队列来削峰。Canal Server 的核心配置在conf/canal.properties里常用参数如下canal.port11111 canal.destinationsexample canal.instance.mysql.slaveId1234 canal.instance.master.address127.0.0.1:3306 canal.instance.dbUsernamecanal canal.instance.dbPasswordyour_password canal.instance.connectionCharsetUTF-8 canal.instance.filter.regextest_db\\..*这里有两个容易被忽略的点canal.instance.mysql.slaveId不能与 MySQL 现有的从库 slaveId 冲突。canal.instance.filter.regex的转义规则很容易看晕。默认写.*表示监听所有库如果只想同步某个业务库需要写成dbname\\..*这里的双反斜杠是 Java 字符串转义的产物别少写了。如果你用的是canal-adapter直接同步到 ES还需要在 adapter 的application.yml里配置数据源和客户端连接。但更通用的做法是把 Canal 接到自己的消费者程序里这样对事件的转换、过滤和失败重试都更有掌控力。后面会详细展开。2.3 索引 mapping 的设计先于同步同步之前先把 ES 索引的 mapping 设计好这件事比写同步代码更重要。如果 mapping 没设计好就急着灌数据后面字段类型一旦错了重新建索引、reindex 的代价非常大。比如 MySQL 订单表里有两个字段created_at是DATETIME金额amount是DECIMAL(10,2)。ES mapping 里的对应关系大概率是{ mappings: { properties: { order_id: { type: keyword }, user_name: { type: keyword }, product_name: { type: text, fields: { keyword: { type: keyword } } }, created_at: { type: date, format: yyyy-MM-dd HH:mm:ss }, amount: { type: scaled_float, scaling_factor: 100 } } } }有几个常见误区order_id这种主键通常要用keyword不要用long因为很多主键其实是字符串或超过 JS 安全整数范围的大整数。金额不要用double精度会出问题。建议用scaled_float把两位小数放大成整数存储。日期格式要统一MySQL 返回的yyyy-MM-dd HH:mm:ss和 JavaLocalDateTime默认序列化格式一旦不一致ES 解析就会失败。mapping 设计完之后再回到同步链路本身就会有清晰的目标我们要把 MySQL 的一行行数据精确转换成上面这套文档结构。3. 核心链路拆解从 binlog 事件到 ES 文档这条链路看起来环节多其实主线非常清晰MySQL 产生 binlog → Canal 伪装成从库拉取解析 → 下游消费事件 → 构造 ES 文档写入索引。下面把中间每一步的原理讲透。3.1 Canal 如何伪装成 MySQL 从库拉取 binlogCanal 的核心机制是把自己伪装成一个 MySQL 的 slave向 master 发送复制协议请求。MySQL 认为它是一个合法的从库就像普通主从复制一样把 binlog 源源不断推送给它。这里要理解一个概念binlog 不是一个大文件而是一系列事件流每个事件有 position偏移量。Canal 会记录自己消费到哪个位置也就是对应的binlog filename和position。这样一旦进程重启它可以从上次记录的位置继续消费而不是从头再来。这个机制带来一个非常重要的运维习惯把 MongoDB、Canal 自己的 position 持久化处理好。Canal Server 默认会把 position 存在本地文件里如果部署在容器里这个文件所在的目录必须做成持久化卷。否则容器重启后 position 丢了要么从头消费要么跳过一堆事件。3.2 三种数据变更事件与 ES 端的对应操作MySQL 的增删改在 binlog 里对应三类事件INSERT、UPDATE、DELETE。Canal 会把它们包装成一个统一的CanalEntry.Entry里面包含操作类型、表名、变更前后的数据。在我们的同步程序里处理逻辑一般是这样的收到INSERT事件就用变更后的整行数据构造 ES 文档执行IndexRequest相当于新增。收到UPDATE事件通常也是执行IndexRequest用变更后的完整行覆盖旧文档。这里很多教程只说更新实际上 doc 的写入是覆盖式的因为我们没有增量字段的概念。收到DELETE事件执行DeleteRequest按主键删掉文档。代码结构大致长这样if (entry.getEventType() EventType.INSERT || entry.getEventType() EventType.UPDATE) { // 取 after columns构造 map MapString, Object doc buildDoc(entry.getAfterColumnsList()); indexRequest new IndexRequest(indexName).id(String.valueOf(doc.get(id))).source(doc); bulk.add(indexRequest); } else if (entry.getEventType() EventType.DELETE) { DeleteRequest deleteRequest new DeleteRequest(indexName).id(String.valueOf(primaryKeyValue)); bulk.add(deleteRequest); }这里有一个关键点**UPDATE事件里拿到的 after 镜像是否完整取决于binlog_row_image是否设置为FULL。**如果设置成MINIMAL可能只包含主键和被修改的字段下游想构造完整文档就会缺字段。这也是前面强调binlog_row_imageFULL的原因。3.3 字段映射规则类型对齐和嵌套结构MySQL 和 ES 的字段类型并不是一一对应的。最大的坑集中在时间、金额和布尔值这三个类型上。MySQL 的DATETIME、TIMESTAMP在 Java 里往往解析成LocalDateTime与 ES 的date类型对齐时要显式指定格式。MySQL 的TINYINT(1)通常代表布尔值但 ES 里如果映射成boolean写入时值又是0/1而不是true/false就会报类型错误。MySQL 的DECIMAL如果直接当成浮点数转成double精度会丢失。所以消费端在构造 ES 文档时至少要做一个轻量的类型转换层类似if (column.getSqlType() Types.DECIMAL || column.getSqlType() Types.NUMERIC) { doc.put(column.getName(), Double.parseDouble(column.getValue()) * 100); } else if (column.getSqlType() Types.TINYINT column.getName().endsWith(_flag)) { doc.put(column.getName(), 1.equals(column.getValue())); } else { doc.put(column.getName(), column.getValue()); }还要考虑嵌套结构。比如一个订单里包含多个商品明细MySQL 里是两张表订单主表和订单明细表。如果 ES 的索引设计成嵌套文档同步的时候就要在消费端做一次明细聚合——把同一订单的所有明细查出来拼成一个数组再写入。这一步不能靠 Canal 自动完成因为 Canal 只做 binlog 解析不做跨表聚合。我见过不少项目在一开始把 ES 索引设计成和 MySQL 表一一对应结果查询的时候还是得在 ES 里搞 join 或者多次查询完全没有发挥出 ES 的优势。所以做字段映射前先想清楚这张索引到底要支撑什么搜索场景然后决定是宽表冗余还是嵌套结构。4. 实操步骤把整条同步链路从零跑通理论讲了不少下面进入动手环节。这一节我会按先通一条链路再扩展的原则来操作先让 Canal 解析出一个表的变化再写一个消费者程序把变化写入 ES最后处理存量数据。4.1 第一步配置 Canal Server 并查看解析日志Canal Server 装好之后启动时可以先跑到控制台看日志bin/startup.sh如果启动失败优先看logs/example/example.log里面会写明连接 MySQL 失败的原因一般有这几种可能账号权限不对、server_id冲突、MySQL 没开 binlog。没有日志信息时用show master status;手动确认 binlog 是否真的在写SHOW MASTER STATUS;正常情况会返回类似mysql-bin.000001 | 12345 | ...。如果返回空说明 binlog 没有开启先回 2.1 节改配置。等 Canal 正常启动后可以用自带的命令行客户端canal-client或者直接写一个最小的 Java 程序抓一条事件确认解析链路是通的CanalConnector connector CanalConnectors.newSingleConnector( new InetSocketAddress(127.0.0.1, 11111), example, , ); connector.connect(); connector.subscribe(test_db\\.order_info); while (true) { Message message connector.getWithoutAck(100); // 打印事件结构 for (Entry entry : message.getEntries()) { System.out.println(entry); } connector.ack(message.getId()); }打印出来的 entry 里能看到 databaseName、tableName、eventType、afterColumns 等字段。走到这一步就说明 Canal 已经把 MySQL 的变更成功变成事件了。4.2 第二步消费者程序写入 ES事件能拿到之后写 ES 就只是 BULK API 的活了。推荐使用 BULK 批量写入不要一条一条 IndexRequest 提交效率差很多。批量写入的核心逻辑BulkRequest bulkRequest new BulkRequest(); // 攒够 1000 条或者 200ms 强制 flush 一次 if (bulkRequest.numberOfActions() 1000) { bulkRequest.timeout(TimeValue.timeValueSeconds(10)); bulkResponse restHighLevelClient.bulk(bulkRequest, RequestOptions.DEFAULT); bulkRequest new BulkRequest(); }这里有两个调优参数值得注意BULK 条数一般从 500 到 2000 之间测太小写入吞吐上不去太大容易把 ES 打出一个TOO_MANY_REQUESTS反而触发背压。单个请求的数据量建议控制在 5MB 到 15MB 之间。如果单条文档很大比如有 nested 数组条数就要调小。写入失败的情况一定要记录并重试。ES 的 BULK 响应里每个子项都有status和error要专门处理409版本的冲突、400的 mapping 不匹配、429的限流。遇到400说明文档结构有问题及时打印完整报错别吞掉。4.3 第三步存量数据的全量初始化增量同步只解决从运行时刻开始的新变更但既有数据怎么办答案是要做一次全量初始化。全量初始化的策略有三种常见路径让canal-adapter的canal.acl配置开启syncModeall配合table字段做全量。但这种方式依赖 adapter且灵活性有限。用数据同步工具如 DataX 把 MySQL 存量数据导出来写入 ES。自己写一个全量脚本分页读取 MySQL 表走和增量同样的字段映射逻辑BULK 写入 ES。然后再启动 Canal 的增量任务两者在同一个主键维度下是幂等的后写入的覆盖先写入的即可。我自己的做法是先用脚本全量灌入再启动 Canal 增量。这里有个先后顺序的问题要注意全量灌入期间如果业务库还在持续变更就可能出现旧数据覆盖新数据的情况。稳妥的做法是记录当前 MySQL 的 binlog position。用这个 position 启动 Canal 增量任务。与此同时做全量导入。全量导入结束后增量消费已经从当时的 position 开始还在持续同步后来的变更。因为 ES 写入是按主键覆盖的即使增量事件在全量导入之前到达了 ES最后全量数据写入也可能会把新值覆盖成旧值。所以正确顺序是如果对一致性要求高可以在某个低峰窗口先停止业务写入记录 position做全量再开增量。如果不能停机那就要用全量数据版本晚于增量事件的方案比如在全量导入后的每条文档上带一个sync_version字段增量写入时判断版本是否比当前新再决定是否覆盖。这个思路在实践里更实用。5. 踩坑实录增量同步最常见的四个坑和排查链路链路跑通不算完生产环境最考验人的是遇到问题时怎么定位。这里把我几次踩坑的完整过程写下来可以照着排查。5.1 主键冲突与 doc id 设计ES 的文档 id 设计如果不合理就会出现同一个业务主键对应多条文档每次更新都产生新文档的问题导致数据重复。我遇到过一次同步用户信息时导出的数据里把文档 id 设成了自增主键但由于消费程序处理重试同一个用户被写入两次ES 里就出现了两个 id 不同的文档查询结果出现重复数据。排查链路先在 ES 里按业务主键做一次聚合统计重复数。再用term查询确认重复文档的_id是什么。最后检查消费程序的 doc id 生成逻辑发现它用了一个自动生成的 UUID而不是业务主键。修复方式很简单把文档_id设定为 MySQL 表的主键值比如order_id。这样 INSERT 和 UPDATE 都走同一个 idES 的写操作天然就是幂等的重复消费也不会产生重复数据。这个原则要在一开始就定死中途修改 id 意味着全量 reindex。5.2 字段类型错乱导致查询结果异常另一个经典坑同步完成之后keyword 字段查不到date 字段排序错乱range 查询返回空。我遇到过的情况是这样MySQL 表里有个status字段是VARCHAR(10)值有WAIT_PAY、PAID、SHIPPED。第一次 mapping 里该字段被 ES 自动推断成了text类型而text类型不参与精确匹配term查询怎么也查不到。更迷惑的是后台页面有的地方能查到数据有的地方查不到因为有的查询条件走了match有的走了term。排查链路用GET /index/_mapping查看映射发现status确实是text。再检查同步日志发现写入时没有任何错误提示因为 ES 对text字段写入字符串是允许的。修复方案把status改成keyword同步代码不变重建索引后全量刷新数据。这告诉我们一个道理写 mapping 时所有用于精确过滤、排序、聚合的字段都必须显式声明为keyword不能靠 ES 自动推断。自动推断只适合日志场景不适合业务索引。5.3 大事务批量更新引发的延迟雪崩Canal 本身很轻但如果你在业务库执行了一个UPDATE ... WHERE create_time 2024-01-01的批量变更一下子更新了几十万行MySQL 的 binlog 会产生几十万个事件。Canal 解析没问题但如果下游是单线程消费BULK 写入的速度跟不上事件产生速度延迟就会越积越高。我当时遇到的情况是业务系统做了一次数据订正凌晨跑了一个大事务结果到早上搜索页的数据还在缓慢更新中用户反馈新增订单半小时后才搜得到。排查链路先看 Canal 的消费延迟指标确认事件积压量。再看 ES 的写入速率发现 BULK 数据量没问题但消费者是单线程串行处理。定位根因消费端从 Canal 拉取事件是单线程的而且每拿到一批事件后还有一次网络请求去查 MySQL 补全订单明细字段遇到大事务就全部排队。修复方案分两层方案一消费端增加线程池把拉事件和写 ES解耦用多个线程并行处理不同表的数据。方案二对同一张表的事件也不一定非要严格串行。因为同一主键的事件如果乱序最终写入的还是同一文档只要最后一次写的是最新值即可。所以可以按主键做哈希后放进不同线程保证同一文档的并发写由 ES 的乐观锁版本控制来兜底。延迟问题一般不会只有一个原因排查顺序建议是先看 Canal 节点本身有没有出现position不前进的情况再看下游消费线程是否阻塞最后看 ES 的写入延迟和 I/O 压力。5.4 sync 任务无故停摆的处理思路还有一种情况任务跑着跑着就停了日志里没有明显的报错数据不再更新。我排查过一次最终发现是 MySQL 的 binlog 到期被删掉了。因为数据库侧的expire_logs_days只配置了一天而 Canal 消费端在某个时间点后停机了几个小时重启后尝试从之前的 position 继续消费但那个 binlog 文件已经被清理MySQL 直接返回了一个错误Canal 进程反复启动失败。排查链路查 Canal 日志能看到类似 binlog not found 或者 position 超出范围的报错。查 MySQL 的SHOW BINARY LOGS;确认最新的 binlog 文件列表。对比发现Canal 记录的 position 指向的文件已经不存在了。修复方案有两个方向方案一如果数据可接受短暂丢失直接把 Canal 的目的地配置清掉让它从当前最新的 binlog position 开始消费再手动对丢失窗口做一次全量补偿。方案二加大expire_logs_days到 7 天甚至更长同时给 Canal 配置监控告警当 position 长时间不前进时及时收到通知。这类问题能暴露的地方很多不只是 binlog 清理。比如磁盘写满导致 position 文件没有落盘成功、容器重启把本地 position 文件冲掉都可能导致重启后衔接不上所以要养成定期检查 position 与告警的习惯。5.5 动态加列时的兼容处理业务表加字段是家常便饭但同步链路的配置很可能跟不上。尤其是线上在 MySQL 执行了ALTER TABLE ADD COLUMN之后ES 的 mapping 里根本没有这个新字段Canal 解析出的 after 镜像里有但消费程序在构造文档时可能忽略它也可能因为strict动态映射策略直接写失败。最佳做法是在消费程序里对未知字段做宽容处理先把所有字段放进 map再用统一的映射逻辑转换。同时在 ES 侧设置dynamic: false而不是strict这样新字段不会导致写入失败但也不会被写入_source之外的结构里。这样即使业务方临时加列也不至于让同步链路整体挂掉。6. 这条链路后续还能怎么扩展一套 Canal ES 的同步链路稳定运行之后你会发现它其实是很多功能的底座不只是搜索。顺着往下走还有几个方向可以参考。多集群同步与跨环境复制。既然 binlog 是同一份只要 Canal 的消费端指向不同环境就能做到同一份数据同时写入测试环境和生产环境方便开发联调时拥有真实数据。接入消息队列削峰。当 Canal 的消费端压力很大时可以把 Canal 输出到消息中间件由消费者按自己的节奏消费这样同步链路和业务系统之间的耦合可以进一步降低。构建数据仓库的实时管道。不止 ES数据仓库也可以订阅同一份 binlog把变更记录落到数仓里做实时指标计算。对事件做审计与回放。因为 binlog 本身就是一份按时间排序的变更日志保留它等于保留了业务表的历史变化轨迹可以用于排查数据问题、做审计甚至做数据回放。在我自己的实践里最实用的扩展其实是把 Canal 的事件格式统一包装成公司内部的消息规范这样下游无论是 ES、数据仓库还是其他存储都能以相同的方式消费数据。这个改造虽然初期要花些功夫但长期会带来很大的便利。回到最初的问题——MySQL 数据同步到 ES核心不是把数据搬过去就结束而是要让两个存储引擎各司其职MySQL 继续做事务处理和强一致存储ES 负责搜索、聚合和筛选。Canal 只是中间那条稳定的流水线真正决定数据质量、查询体验的还是 mapping 设计、字段映射逻辑和消费程序的稳定性。这几个方面做到位这套方案跑几年都不会有大问题。