1. 先聊清楚这套方案到底解决什么问题1.1 为什么要把MySQL的数据搬到ES先说个我实际遇到的场景。某个后台系统的订单表业务跑了两三年以后单表过了千万级用户在前台按订单号、客户姓名、手机号、商品名称搜历史订单赶上一个奇怪的关键词一条深分页的模糊查询能直接把数据库CPU打到80%以上。加索引like查询一旦带了前导通配符常规B Tree索引直接失效只能硬扫。后来我们决定上ElasticSearch用倒排索引来扛这种任意字段组合检索的场景。这个诉求本身不新鲜MySQL擅长事务性写入ES擅长全文检索和复杂过滤聚合。问题是数据不能靠人肉搬运也不能让业务系统在每次增删改的时候顺手往ES里也写一份。业务方代码里双写等于把同步逻辑和一个高可用中间件绑定到业务事务里面一方失败两边就出现脏数据而且排查成本极高。所以基于binlog做增量数据同步就成了最稳妥的路子。把MySQL的变更日志实时采集出来翻译成下游能听懂的消息再交给ES消费写入。这条路子不需要业务代码做任何改造对线上系统零侵入这就是Canal这类CDCChange Data Capture变更数据捕获组件存在的意义。1.2 中间为什么非要加一个Kafka很多人在初期会有一个疑问Canal不是可以直接把数据推到ES吗为什么中间还要再堆一个Kafka平白多了一套要运维的东西。直接推当然能做Canal本身确实有输出到ES或者直接HTTP回调的能力但我在实际落地时几乎一定会选择中间加消息队列。原因有几条第一是削峰填谷。电商大促或者业务高峰的时候数据库的变更量会突然暴涨而ES集群的写入能力是有上限的。Canal直接推ESES扛不住的时候Canal也被拖死甚至会倒逼binlog拉取积压。中间加一个Kafka相当于给整条链路装了一个巨大的缓冲池ES慢一点没关系消息先落在Kafka里面消费端慢慢追。第二是一对多的分发。一份binlog变更往往不只是给ES用。同步到数仓做统计分析、刷新缓存、构建下游宽表这些都是常见的需求。只要Canal把消息投递到Kafka任何一个下游都可以通过独立的消费组去消费同一份数据互不干扰。第三是可回溯和容错。Kafka消息有保留时间消费端哪怕崩了一整天只要offset没丢起来之后从上次位置继续消费就能把数据补齐。这个能力在跨团队协作场景下非常实用。用个生活化的类比Canal像一个专职盯梢的观察员只负责把看到的“变化”记录下来Kafka就像一个蓄水池既承接上游的水量也按需放水给下游多个工厂。没有这个池子上游稍微抖一下下游就要跟着抖。1.3 整条链路长什么样我落地这套方案之后经常见人问架构长什么样。用文字描述就是MySQL开启binlog且格式为ROWCanal伪装成一个MySQL从库去订阅binlog把解析后的增量变更事件投递到Kafka的指定Topic自研的消费端服务从Kafka拉取消息经过解析、字段转换、幂等处理后通过ES的Bulk接口批量写入索引。把链路各环节拆开看职责非常清晰环节组件核心职责数据源MySQL业务数据的真实来源开启binlog作为变更依据变更采集Canal伪装成MySQL从库订阅binlog并解析成结构化消息缓冲传输Kafka削峰、隔离、消息保留、一对多分发消息消费自研同步服务拉取消息解析事件类型构造ES文档批量写入目标存储ElasticSearch提供倒排索引与聚合检索能力这篇文章接下来的内容就是从环境准备、Canal配置、消息格式解析、消费端代码到生产环境常见故障排查的一条龙落地记录。想看具体代码的可以直接跳到第4节但我强烈建议先把前面的配置讲透因为后面90%的坑都在配置和消息格式上。2. 动手前的环境准备这三样东西一个都别漏2.1 MySQL端不把binlog开对后面全是白搭很多人装好Canal之后启动日志一切正常但数据库改了数据就是收不到消息。第一个排查点永远是MySQL的binlog配置。Canal工作的前提是MySQL开启了binlog并且格式必须是ROW。为什么必须是ROW这要从binlog的三种格式说起。STATEMENT格式记录的是原始SQL语句比如UPDATE orders SET status2 WHERE id1001。对Canal来说它只能拿到这句SQL没法知道这条记录更新之前是什么值、更新之后受影响的具体行有哪些更没办法直接推算出一个完整的JSON消息。而ROW格式记录的是每一行变更的前后镜像Canal才能把变化拆成一个个结构化事件。MIXED格式呢某些场景下会退化成STATEMENT同样不可靠。另外还有一个容易被忽略的参数binlog_row_image。这个参数决定了ROW模式下记录的字段范围。生产环境必须设置成FULL否则默认的MINIMAL模式只记录发生变化的字段和主键消费端拿不到整行数据后面做ES全量覆盖写入的时候就会遇到大麻烦。建议的MySQL配置如下[mysqld] server-id 1 log-bin mysql-bin binlog_format ROW binlog_row_image FULL expire_logs_days 7 max_binlog_size 256M其中server-id必须设置而且不能和Canal伪装的从库ID冲突否则MySQL主从复制协议会直接报错。改完配置记得重启MySQL实例才能生效。然后创建一个专用的Canal账号不要用业务账号去订阅binlog权限最小化是原则CREATE USER canal% IDENTIFIED BY canal_pass; GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO canal%; FLUSH PRIVILEGES;这几个权限的用途SELECT权限是让Canal读取表结构做字段映射REPLICATION SLAVE权限是让它以从库身份拉取binlog日志流REPLICATION CLIENT权限是让它能执行SHOW MASTER STATUS这类命令定位当前位点。没有这些权限Canal会一直处于连接失败的状态日志报错也容易误导人。配置完可以用下面两条SQL自查SHOW VARIABLES LIKE binlog_format; SHOW VARIABLES LIKE binlog_row_image;2.2 Kafka和Topic的基本设计Kafka在这一条链路里的定位是缓冲和分发Topic的规划直接决定了消费端代码的复杂度。我的建议是按照业务线来建Topic而不是一个表一个Topic。比如订单库建一个canal-order商品库建一个canal-product。好处是消费者只需要订阅有限的几个Topic而且同一个Topic里可以承载多个表的消息消息体内自带database和table字段消费端可以内部路由。另外一点一个实例的Canal配置里可以让每个instance任务分别对应不同的Topic。比如订单库的instance产生的消息进canal-order商品库的instance进canal-product。这样互不影响某个Topic积压了也能独立调优。Topic的分区数怎么定我给一个通用的粗算思路先预估业务峰值时刻每秒产生的binlog变更消息量比如每秒5000条再估算单分区的消费能力按一个消费者每秒能处理2000~5000条来估当然这个数字取决于消费端逻辑重不重。要留出弹性那么分区数设为3到5个比较合适。如果只同步一个索引且数据量不大也可以直接从2到3个分区起步后续不够再加分区。但要注意Kafka不推荐一个Topic只有1个分区那样消费并发能力就被锁死了Canal投递消息的性能也会受限。消息保留时间建议设置成3到7天。同步链路偶尔出问题运维希望的通常是能直接回放消息修复而不是重新全量导数据保留时间长一点处理问题的余地就大很多。2.3 ES索引的mapping真的要先建好我见过不少团队Canal跑起来了消息也进来了才发现ES索引字段类型全是text数字比较、范围查询通通不对然后又要重建索引。这种事一定要在同步开始之前就避免。ES的mapping一旦创建大部分字段类型是无法直接修改的。在Canal消息进入消费端之前就要先设计好索引结构。比如订单索引金额用scaled_float或double状态用keyword订单时间用date商品名称用text并配ik_max_word分词器。注意不要把所有字段都设成text否则你之后做精确过滤、聚合统计、排序都会很痛苦。分片数的估算给个原则单个分片的数据量建议控制在20~50GB以内按最终一年的数据量反推分片数。副本数按读多写少的需求来一般1个副本就够了。索引别名也建议加上。线上需要调整mapping或者滚动索引时可以平滑切换不用停同步服务。这块准备工作做好后面才能专心调Canal和消费端。3. Canal配置要点从全局到实例3.1 canal.properties里影响链路的关键项Canal部署好之后第一个要改的是conf/canal.properties这个全局配置文件。里面接近上百个配置项但真正决定Kafka投递行为的就是几个。# 指定模式为kafka canal.serverMode kafka # Kafka连接地址集群就写多个逗号分隔 canal.mq.servers kafka1:9092,kafka2:9092,kafka3:9092 # 是否投递扁平JSON消息必须为true canal.mq.flatMessage true # 设置投递到Kafka的默认Topic实例配置可以覆盖 canal.mq.defaultTopic canal-default # 并行发送线程数 canal.mq.parallelThreadSize 16canal.mq.flatMessage这个配置尤其重要。它为false时Canal投递给Kafka的是Protobuf序列化的二进制字节流消费者要用Canal提供的客户端库反序列化才能读到内容调试成本很高。为true时Canal会直接把事件转成可读的JSON数组投出去。我建议所有场景都开flatMessage除非你的下游就是Canal的官方client在做二次开发。3.2 instance.properties把同步目标锁死全局配置之外Canal的每一个同步任务对应一个instance目录默认在conf/example下。真正的同步范围配置都在instance.properties里面。# MySQL实例地址 canal.instance.master.address 10.0.0.10:3306 # 上面创建的canal账号 canal.instance.dbUsername canal canal.instance.dbPassword canal_pass canal.instance.connectionCharset UTF-8 # 要订阅的库表格式是 库名\\.表名 canal.instance.filter.regex mall\\.orders,mall\\.users # 黑名单排除不需要的表 canal.instance.filter.black.regex mall\\.t_temp.*,mall\\.t_log.* # 这个实例投递到的Topic覆盖全局默认值 canal.mq.topic canal-order这里正则的写法是新手最容易翻车的地方。mall\\.orders中的两个反斜杠是必须的因为整个配置值会被Canal用正则表达式去解析而正则中匹配一个点号.必须转义成\\.。很多人在这个字符串上反复出问题漏一个反斜杠就发现Canal一条消息都不报。正则支持通配符。mall\\..*表示mall库下所有表.*\\..*表示所有库所有表。黑名单的优先级高于白名单也就是正则匹配的表会先被过滤掉。比如你白名单里配了mall\\..*黑名单里写mall\\.t_temp.*那么mall库下所有以t_temp开头的表都会被排除。3.3 启动Canal并验证消息流出启动和验证的流程很简单但每个细节都值得确认一次。进入Canal安装目录执行bin/startup.sh然后看日志确认没报错tail -f logs/canal/canal.log tail -f logs/example/example.logexample.log中如果出现connected to字样说明Canal已经成功连接到了MySQL。如果出现权限错误或者连接超时先回头检查Canal账号权限和MySQL实例地址。验证链路通不通最快的方式是开一个Kafka终端消费者订阅Canal投递的那个Topickafka-console-consumer.sh --bootstrap-server kafka1:9092 --topic canal-order --from-beginning然后去数据库里手动执行一条UPDATE或者INSERT如果终端能看到JSON消息输出说明从binlog采集到Kafka投递这一段已经通了。这是我个人习惯的验证顺序先验MySQL到Kafka再写消费端这样出了问题不会被上下游互相干扰的假象误导。4. 消费端把消息吃进去并同步到ES4.1 先看懂Canal投过来的JSON在动手写代码前一定要先看懂Canal的flatMessage格式。它投递到Kafka的是一条JSON数组数组里的每个元素代表一个变更事件。这里以一个订单表的UPDATE为例[ { id: 560, database: mall, table: orders, isDdl: false, type: UPDATE, data: [ { id: 1001, order_no: ORD20240501001, status: 2, pay_time: 2024-05-01 12:30:00, total_amount: 199.00, update_time: 2024-05-01 12:30:05 } ], old: [ { status: 1, pay_time: null } ] } ]几个关键字段的含义type事件类型常见的有INSERT、UPDATE、DELETE还有CREATE、ALTER这些DDL类型。data变更后的数据数组。注意数组里每个字段的值在Canal的JSON里全是字符串因为binlog里面存的就是字符串表达。old变更前的数据数组只包含发生变化的列。对于UPDATE事件data里是完整的更新后镜像old里是发生变化的字段的旧值。isDdl是否为DDL语句同步到ES的一般直接过滤掉。database和table标明这条消息来自哪个库哪张表消费端可以根据这个字段做表到索引的路由。为什么前面反复强调binlog_row_imageFULL就是因为FULL模式下data里才是整行的完整快照消费者拿data直接就能拼出一个完整的ES文档。如果是MINIMAL模式data里可能只包含变化的列和主键那消费端要做字段合并才能写ES复杂度和出错率都会上去。还有一个细节值得注意Canal投递的消息是一次性投递一个数组这个数组可能包含同一个事务里的多个事件。消费端处理时不要只取message[0]要遍历整个集合。4.2 三类事件的同步逻辑怎么定消费端最核心的就是事件类型的分流逻辑我按三种类型分别处理INSERT事件。data[0]就是新增记录的全部字段直接构造ES文档并指定主键用IndexRequest写入索引。UPDATE事件。这是最需要抠细节的类型。绝大多数情况下主键没有变化那就直接用data[0]整行覆盖写入ES文档。之所以用整行覆盖而不是用UpdateRequest做部分字段更新是因为data里本身就是FULL镜像包含所有字段覆盖写入既简单又能保证ES文档和MySQL的记录完全一致。用UpdateRequest做部分更新万一漏字段或者字段名映射错了ES里就会残留旧值而且这类问题很难一眼看出来。但有一种边界情况UPDATE事件里data和old的主键字段值不一样说明这条记录的主键被修改了。MySQL主键变更虽然在业务规范里应该杜绝但真碰到了消费端必须做两步操作先按old里的旧主键删除ES里的旧文档再按data里的新主键写入新文档。不然ES里会同时存在新旧两条数据。DELETE事件。MySQL删除一条记录时Canal投递的data里是删除前的整行快照消费端直接从data[0]里取主键字段调用DeleteRequest删除ES文档。DDL事件比如ALTER TABLE、CREATE TABLE直接忽略。如果业务上确实有关联需求可以加日志留痕但默认不要影响主流程。4.3 核心代码解析、转换、批量写入下面这段是我实际用过的消费端核心流程基于Spring Boot加Spring KafkaES客户端以ElasticSearch 7.x为例。代码做了简化但保留了主干结构。Component Slf4j public class CanalEsSyncConsumer { private static final ObjectMapper MAPPER new ObjectMapper(); Resource private RestHighLevelClient esClient; KafkaListener(topics canal-order, groupId es-sync-order-group) public void onMessage(String message) { try { ListCanalMessage eventList MAPPER.readValue(message, new TypeReferenceListCanalMessage() {}); BulkRequest bulkRequest new BulkRequest(); for (CanalMessage event : eventList) { if (event.getIsDdl()) { continue; } if (event.getData() null || event.getData().isEmpty()) { continue; } buildBulkRequest(bulkRequest, event); } if (!bulkRequest.requests().isEmpty()) { BulkResponse response esClient.bulk(bulkRequest, RequestOptions.DEFAULT); if (response.hasFailures()) { log.error(ES bulk写入存在失败项: {}, response.buildFailureMessage()); // 这里根据业务需要决定抛异常回滚消费位点还是走补偿表 } } } catch (Exception e) { log.error(处理Canal消息失败, e); // 抛异常触发Kafka重试 throw new RuntimeException(e); } } private void buildBulkRequest(BulkRequest bulkRequest, CanalMessage event) { MapString, Object data event.getData().get(0); String primaryKey String.valueOf(data.get(id)); String indexName routeIndex(event.getTable()); switch (event.getType()) { case INSERT: case UPDATE: if (isPrimaryKeyChanged(event)) { // 主键变更先删旧文档再写新文档 MapString, Object old event.getOld().get(0); String oldId String.valueOf(old.get(id)); bulkRequest.add(new DeleteRequest(indexName).id(oldId)); } IndexRequest indexRequest new IndexRequest(indexName) .id(primaryKey) .source(data); bulkRequest.add(indexRequest); break; case DELETE: bulkRequest.add(new DeleteRequest(indexName).id(primaryKey)); break; default: break; } } private String routeIndex(String table) { if (orders.equals(table)) { return orders_index; } return default_index; } private boolean isPrimaryKeyChanged(CanalMessage event) { if (event.getOld() null || event.getOld().isEmpty()) { return false; } MapString, Object old event.getOld().get(0); MapString, Object data event.getData().get(0); Object oldId old.get(id); Object newId data.get(id); return oldId ! null newId ! null !oldId.toString().equals(newId.toString()); } }这段代码里有几个点要重点说明。Bulk批量提交是ES写入的核心手段比一条一条写吞吐高出一个数量级。批量大小建议控制在500到1000条之间或者按总字节数控制在5MB左右。开太小了网络往返开销占比高开太大了单次bulk耗时会变长一旦出差错整个批次的重试成本很高。消费端抛异常触发Kafka重试时默认的Spring Kafka机制会导致同一个批次反复消费。所以消费逻辑必须幂等。ES的index写入天然幂等重复执行结果一致这块可以放心。CanalMessage这个对象我直接用一个POJO承载字段包括type、database、table、isDdl、data、olddata和old用ListMapString, Object表示。这种写法比硬编码字段名灵活后续表结构加字段消费端代码都不用改。4.4 顺序性、幂等性与并发注意这里要单独把顺序性拎出来讲因为它直接决定数据对不对。Kafka只能保证单个分区内的消息是有序的。如果同一行数据的主键操作分散到了不同分区消费端并行处理时完全可能出现旧消息后到、覆盖新数据的情况。举个例子一个订单先是UPDATE把status改成2紧接着又UPDATE改成3如果这两条消息被投到不同分区消费端并发执行结果可能是status为2的旧消息最后写入ES里永远停在2。Canal提供了一个配置来按主键hash投递分区非常关键# 表示mall库orders表按id字段hash决定投递到哪个分区 canal.mq.partitionHash mall.orders:id这样同一个订单id的所有变更消息都会进入同一个分区消费端只要保证每个分区单线程消费顺序就保住了。如果还需要再拆分线程提高吞吐可以在消费端内部按主键再做一次hash分桶把同一主键的消息分到同一个本地内存队列里由对应的单线程处理。本质上就是做一个两级分区一级由Kafka保证一级由消费端内部自建。幂等性其实要分两层看。ES层面的index写入天然幂等重复执行没问题但UpdateRequest配合doc做部分更新就不是严格幂等的必须谨慎。我的个人习惯是能走index全量覆盖就不走update流程简单得多。还有消费者实例数。多个消费者实例属于同一个groupId订阅同一个Topic时Kafka会把分区分摊给这些实例。消费者实例数不要超过分区数否则多出来的实例一直是空闲状态还会引起无意义的rebalance。这个点排查消息积压时经常遇到。5. 上线后我踩过的坑排查与调优实录5.1 数据库改了数据Kafka却一直没消息这个问题排在所有问题的第一位遇到概率最高。从下往上排查我一般按这个顺序过一遍先查MySQL的binlog配置。SHOW VARIABLES LIKE binlog_format很多库默认是MIXED或STATEMENTCanal采集不到可靠的行级数据表现为日志显示连接正常但没有任何消息。这个检查是零成本的建议第一时间做。再查instance里的filter.regex正则。最常见的翻车点是漏了反斜杠。mall\\.orders和mall.orders就差一个反斜杠前者正常匹配后者正则里的点号会匹配任意字符看起来像配置了对实际上匹配到的表和预期完全不同甚至直接什么都没匹配上。然后查Canal账号权限。如果没有REPLICATION SLAVE权限Canal拉取binlog时会报Access denied但有些版本的Canal日志里只显示一条连接异常很容易被忽略。最后查黑名单。如果某个表被filter.black.regex命中了Canal会静默过滤掉日志不报任何错误。5.2 UPDATE之后ES里还是旧数据这个问题的优先级也很高。数据改了消息也收到了但ES里查出来还是旧值。我的排查路径有这几条消费端程序是不是把UPDATE事件漏掉了。检查代码里switch的case分支很多人的case只写了INSERT和DELETE忘了UPDATE。Canal投递的type字段值是大写的匹配时大小写敏感也是一个隐藏雷点。消费组是不是有多个。如果启动了多个消费端服务实例但groupId配置不一致就会出现两个服务同时消费同一个Topic的情况其中一份逻辑是有问题的旧版代码两边互相覆盖数据看起来就忽新忽旧。还有一类罕见但真实存在的问题MySQL里记录更新了但update_time在业务代码里没有更新看起来像没同步。这种其实不是同步链路的问题是源端数据本身就没变。如果确认消息和代码都没问题再查ES的refresh机制。ES默认refresh间隔是1秒写入后立即检索可能查不到最新数据。这个不算故障但第一次接触ES的人经常会误判。5.3 主键变更和无主键表的边界情况ES文档必须有唯一的_id这个_id我们通常直接取MySQL主键。一旦MySQL表没有主键Canal解析的时候会直接报错因为binlog事件里没有主键可以追踪行变化。这不是消费端能绕过去的规范上就是要求业务表必须有主键。主键变更的问题在5.3节的代码里已经给出了处理方式。但我想补充一个真实场景很多业务不做真实删除而是软删除加状态标记然后又重新插入一条相同业务含义的记录主键却是新的。这种情况下ES里旧文档还在新文档也写进来了就会出现重复记录。针对这类业务消费端可以考虑在软删除事件里也做一次ES删除或者约定一个逻辑主键来做去重。具体怎么设计取决于业务对数据一致性的要求。5.4 Kafka消息积压与同步延迟过高消息积压的表现是Kafka的consumer lag越来越高ES的新数据迟迟不出现。先不要急着加机器按这几个维度排查消费者实例数是否少于分区数。如果8个分区但只有2个消费者实例在跑那最多只有2个分区被消费剩下6个分区的消息完全停滞。要么加消费者实例要么给消费逻辑提速。单批次消息解析和写入是否超时。有些消费端设计里对每条消息都做一次ES请求效率极低。一定要走Bulk批量写入。Bulk批次调大能显著降低平均耗时。ES集群的写入瓶颈。分片数太少单分片写入成为瓶颈refresh_interval如果设置成1秒写入压力大时会产生大量小Lucene分段合并拖慢整库的写入性能。可以在索引设置里适当调大refresh间隔比如30秒换取更高的写吞吐。消费端本身的HASH分桶线程设计不合理也会导致某几个分桶堆积。这要看代码里是否有热点键的问题比如某个用户的订单量特别大hash到同一个桶其他桶空闲这个桶一直堆积。5.5 延迟监控和日常运维习惯同步链路没有监控等于裸奔。我采用的方案是在ES文档里额外写入一个sync_time字段值是消费端的系统时间。业务表里本来就有update_time如果源表没有这条链路建议推着业务方补上对比ES文档的sync_time和MySQL的update_time两者差值就是这条链路的真实同步延迟。Kafka侧最直接的监控指标是consumer lag也就是log_end_offset减去current_offset。这个值可以接入Prometheus和Grafana配上告警阈值。有一个自定义告警的经验值如果某个Topic的lag持续超过1万条并且没有下降趋势就该人工介入了。日常运维里最实用的习惯是保留Canal的位点信息。Canal的meta信息记录在conf/example/meta.dat里包含上一次消费的binlog文件名和位点。如果观察Kafka消息没问题但数据对不上可以把这个位点文件和MySQL的SHOW MASTER STATUS结果对比能快速判断Canal是从哪个位置开始采集的。还有一个小技巧Canal的安装目录升级时建议先把旧meta.dat文件备份升级完如果Canal异常备份的位点文件可以用于快速回退避免重新从某个时间点开始同步导致大数据量的补数。最后再补一个很多团队忽略的点。整条链路的配置项包括MySQL的binlog参数、Canal的filter正则、Topic分区数、消费端groupId都建议固化到配置管理平台或者至少写进项目的README里。我见过最多次的返工都是因为某个环境用了旧的配置但没人知道排查了大半天才发现是配置漂移造成的。这套组件本身很成熟真正让开发头疼的从来都是环境差异和配置不一致。