1. 同步方案选型为什么你的MySQL和ES之间需要Canal把MySQL业务数据同步到Elasticsearch做检索几乎是每个搜索系统都绕不开的活。网上讲Canal的教程不少但大多停在“能通就行”真正到了生产环境filter配置、位点管理、消费端幂等、全量增量衔接每个环节都能给你上一课。这篇文章就围绕“使用canal同步mysql数据到es”这条主线把服务端配置、客户端消费、ES写入和问题排查完整讲一遍。先说结论如果你的主数据在MySQL检索和报表在ES业务能接受秒级延迟那Canal是增量同步最合理的选择。它能把自己伪装成一个MySQL从库订阅binlog日志拿到每一次行变更再交给下游消费端写入ES或MQ。MySQL不需要改业务代码ES不需要懂业务表结构中间只多了一个增量管道。这套方案适合谁适合三类人第一类是正在做商品、订单、用户这类主数据索引同步的后端开发第二类是让Elasticsearch承担聚合查询、但又不想手工维护双写的运维或数据工程师第三类是看了很多教程但没搞懂“为什么这么配”的初学者。我尽量把每个参数背后的逻辑说清楚而不是只给一份能跑通的配置。1.1 同步需求的本质是什么表面上是把A库的数据搬到B索引本质上要解决三件事捕获、传输、回放。捕获指的是怎么知道MySQL发生了一次更新。业务代码里手动去查变更太被动轮询整张表代价又太大。只有从binlog入手是最自然的方式因为MySQL本身就把每一次插入、更新、删除都以二进制日志的形式记录下来了而且是顺序追加的。Canal就是把这个日志拉出来做解析。传输指的是Canal解析出来的增量数据怎么到ES。可以直接用TCP接口拉取也可以先投到消息队列再消费。中间加不加MQ取决于你后面有几个消费端、需不需要堆积和回溯。小规模场景直接TCP就够了数据量上来之后MQ的缓冲价值才会体现。回放指的是消费端拿到增量事件后怎么作用到ES文档。这一步最容易被忽视但它才是出问题最多的地方。ES不是直接照搬表结构和字段需要确定文档ID、字段映射、类型映射、删除逻辑和幂等规则。很多人Canal配好了增量却写错了基本都是栽在回放这步。1.2 为什么不建议用业务双写和定时任务早期很多系统直接在业务流程里同步ES比如插入MySQL之后立刻调用ES接口。这种方案在小数据量下跑得通但隐患非常大。一旦ES服务抖动业务主线程会被拖慢甚至失败后还要回滚数据库事务和ES写入变成了一个分布式事务问题。加一个字段就得改同步代码漏改一处两边数据就差了一截。定时任务同步也有问题。如果你依赖一个updated_at时间字段做增量查询那么删除操作基本没法发现除非用软删除否则ES里永远是“多出来的文档”。而且像订单表动辄几千万行全表扫描更新时间字段的代价在业务低峰也是不小的压力。定时任务的延迟还不可控一个任务重跑下游就会拿到一段时间内不连续的状态。Canal这类binlog工具走的完全是另一条路它不碰业务代码不会给主流程增加额外延迟也不用关心表里有没有时间字段因为MySQL已经把每一次变更都写下来了。它甚至能拿到变更前后的快照这意味着你可以精确知道某一行“从什么变成了什么”这对回放ES时的更新、删除、宽表打散都非常重要。1.3 Canal的工作过程是怎么样的Canal的工作逻辑并不复杂。它启动后向MySQL注册成为一个“伪从库”发送dump协议请求binlog数据。MySQL会像对待真实从库一样把binlog事件推送给Canal。Canal解析这些事件区分事务边界把RowData转换成通用格式然后通过自定义协议交给客户端或者写入MQ。这个设计有一个天然好处它复用了MySQL成熟的复制机制不会对主库造成额外负担也不需要给MySQL安装任何插件。只要主库开启了ROW模式的binlogCanal就能准确拿到每一行的变更数据。架构上一般是这样MySQL主库 - Canal Server - 消费端Filter/Transformer - Elasticsearch。Canal Server本身可以管理多个instance每个instance对应一组独立的订阅任务有自己的位点、过滤规则和消费进度。这样同一套Canal Server既能给ES同步也能给其他系统同步只要分开instance就行。2. 部署前必须做好的准备MySQL与Canal服务端配置很多人在这一步翻车不是Canal本身有问题而是MySQL侧的基础条件没满足。Canal的原理是把自己伪装成从库拉binlog所以MySQL必须开启binlog而且格式必须是ROW。如果是老库迁移大概率以前为了节省空间只开了STATEMENT或者MIXED这种格式下Canal拿不到完整字段值解析出来的数据会缺字段。2.1 MySQL侧需要调整的配置先看MySQL当前的binlog情况SHOW VARIABLES LIKE log_bin; SHOW VARIABLES LIKE binlog_format; SHOW MASTER STATUS;如果log_bin是OFF那就需要修改MySQL配置文件。以Linux下的my.cnf为例[mysqld] server-id1 log-binmysql-bin binlog-formatROW binlog-row-imageFULL expire_logs_days7 max_binlog_size256M这里有几个坑。server-id必须手动指定不能和现有从库、其他Canal实例、以及任何复制节点的server-id冲突否则MySQL会误判复制拓扑直接把连接踢掉。binlog-row-image最好设置成FULL默认其实也是FULL但如果你为了节省binlog体积改成了MINIMAL那只记录被修改的字段Canal拿到UPDATE的before列就不完整消费端做数据对比和宽表更新时会缺参照。expire_logs_days建议保留7天以上。Canal出现故障或者暂停维护时binlog不会被立刻清理给修复留出空间。如果binlog只保留一天消费端停了三天再启动MySQL已经purge掉对应位点的日志Canal只能重新全量同步。改完配置重启MySQL后创建Canal专用账号CREATE USER canal% IDENTIFIED BY canal; GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO canal%; FLUSH PRIVILEGES;这三个权限分别用来读取数据、模拟从库拉binlog、查询master状态。别图省事用root账号一旦Canal配置或账号密码泄露权限范围太大。2.2 Canal Server安装与快速启动我以Canal 1.1.7版本为例。下载canal.deployer包后解压目录结构大概是bin/启动和停止脚本conf/全局配置和实例配置logs/运行日志lib/依赖包启动前先改全局配置conf/canal.properties。如果是单机直接TCP模式主要确认这几个参数canal.port 11111 canal.serverMode tcp canal.instance.global.mode tcp其中canal.port是客户端连接的端口serverMode指定Canal对外服务方式。我们先用tcp模式如果后续要接Kafka或RocketMQ再改成对应模式。然后进入conf/example/目录这是默认实例名。重点修改instance.propertiescanal.instance.mysql.slaveId 1234 canal.instance.master.address 127.0.0.1:3306 canal.instance.dbUsername canal canal.instance.dbPassword canal canal.instance.connectionCharset UTF-8 canal.instance.filter.regex test_db\\.product canal.instance.filter.black.regex mysql\\.slave_.*mysql.slaveId是Canal伪装的从库ID同样不能和真实从库冲突。master.address填MySQL主库地址如果有多个Canal节点订阅同一个主库每个节点的slaveId都要不同。filter正则看起来简单实际是坑最多的配置。test_db\\.product表示只订阅test_db库下的product表。注意正则里库名和表名之间的点号必须写成\\.因为正则里.是任意字符。如果你写成test_db.product它会匹配到test_dbXproduct这种名字反而漏掉真正的表。想订阅整个库可以写test_db\\..*想订阅所有库所有表可以写.*\\..*。启动服务bin/startup.sh然后看日志确认状态tail -f logs/canal/canal.log tail -f logs/example/example.log正常启动后canal.log里会出现Subscribe相关的记录example.log里会打印拉取binlog的位点。如果这个文件一直报错说明前面某些配置没满足。2.3 几个容易忽视的实例参数除了上面那几个还有几个参数在数据量上来之后非常关键。canal.instance.memory.bufferMemory决定了instance的环形缓冲区大小默认是64KB还是1024MB要看版本生产环境建议根据单条binlog大小调到256MB或512MB。如果缓冲区太小大事务一次性变更几十万行Canal内部会积压报错客户端消费速度跟不上。canal.instance.batch.mode控制批量模式可以设置成MYSQL或MEMORY。MYSQL模式会按MySQL事务边界切分批次能保证一个事务的数据在同一批消息里对事务一致性要求高的场景很有用。MEMORY模式则允许跨事务拼接吞吐更高但下游需要注意事务边界。canal.instance.parser.parallelThreadSize是binlog解析线程数。可以提升解析吞吐但不要盲目调大它会改变分发的顺序性。后面讲消费端时我会专门说乱序问题。3. 消费端实现从Canal拉取消息并写入ES服务端配置完成只是开始真正要花时间的是消费端。Canal自己不做数据同步到ES的动作它把变更数据吐出来之后怎么过滤、怎么转换、怎么写入ES全由下游决定。3.1 两种接入方式怎么选Canal对外主要提供两种方式TCP接口和MQ消息。TCP方式下客户端依赖canal.client这个Java类库直接连到Canal Server的11111端口拉消息。优点是延迟最低、不依赖额外组件适合单消费端、单ES集群的场景。缺点是消费逻辑和客户端生命周期绑死服务端重启后位点恢复需要自己掌握。MQ方式下Canal Server会把解析后的消息投到Kafka、RocketMQ或RabbitMQ消费端从MQ拉取。优点是消息被持久化到MQ消费失败可以重新消费多个系统可以同时订阅同一份变更数据。缺点是多引入了一套MQ的运维成本延迟也略高一点点。如果你只是“同步MySQL到ES”这一个目标我建议先用TCP方式跑通。等到出现多消费端、需要回溯、或者Canal和ES不在同一机房时再上MQ不迟。3.2 Java客户端的基本代码结构引入canal.client依赖后最基础的使用方式是CanalConnector connector CanalConnectors.newSingleConnector( new InetSocketAddress(127.0.0.1, 11111), example, , ); connector.connect(); connector.subscribe(test_db\\.product); while (running) { Message message connector.getWithoutAck(100); long batchId message.getId(); if (batchId -1 || message.getEntries().isEmpty()) { Thread.sleep(200); continue; } ListEntry entries message.getEntries(); handleEntries(entries); connector.ack(batchId); }注意这里用的是getWithoutAck不是get。get方法在调用时就自动确认了消息一旦消费之后还没写入ES程序就崩了这批数据就丢了。getWithoutAck把确认动作留到处理完成之后再执行只有ack成功Canal服务端才会把位点往后推。这是保证不丢数据的核心。subscribe参数可以覆盖服务端实例里的过滤规则相当于在Canal配置之上再加一层过滤。实际开发中最好保持消费端subscribe和服务端filter一致否则两边各自过滤出了问题非常不好排查。处理Entry的时候需要区分一下条目类型TRANSACTIONBEGIN、ROWDATA、TRANSACTIONEND。我们只需要关心ROWDATA。每条ROWDATA里是一个RowChange对象里面携带了事件类型和行数据列表。for (Entry entry : entries) { if (entry.getEntryType() ! EntryType.ROWDATA) { continue; } RowChange rowChange RowChange.parseFrom(entry.getStoreValue()); EventType eventType rowChange.getEventType(); if (eventType EventType.DDL) { continue; // DDL事件一般单独处理 } for (RowData rowData : rowChange.getRowDatasList()) { ListColumn beforeColumns rowData.getBeforeColumnsList(); ListColumn afterColumns rowData.getAfterColumnsList(); // INSERT: afterColumns有值before为空 // UPDATE: before和after都有值 // DELETE: before有值after为空 } }拿到Column列表后一般会构建一个MapMapString, Object dataMap new HashMap(); for (Column column : afterColumns) { if (column.getIsNull()) { dataMap.put(column.getName(), null); } else { dataMap.put(column.getName(), column.getValue()); } }字段名直接取自MySQL列名类型默认是字符串。如果你要精确处理数字类型可以用column.getSqlType()判断JDBC类型再做转换。不过大多数ES映射会自己处理字符串转数值所以这个转换不是必选项。3.3 写入ES的幂等和批量策略ES文档和MySQL行并不是总能一一对应。最简单的情况是一张表对应一个索引那直接把主键映射成_id就行。但实际项目里经常是宽表MySQL里product、category、brand三张表ES里一个product_index索引。这种场景下Canal的三张表变更事件都需要汇入同一个消费逻辑通过关联查询组装成ES文档。Canal只是负责告诉你哪张表变了不负责帮你把宽表拼好。无论哪种模型最关键的一点是ES文档的_id必须显式指定。如果你让ES自动生成_id那么同一条MySQL记录的插入、更新、删除事件会被散落到不同文档里幂等性不复存在。一般做法是直接用主键字段String docId String.valueOf(dataMap.get(product_id));批量写入用BulkRequest不要一条一条调用ES接口。以ES 7.x为例BulkRequest bulkRequest new BulkRequest(); for (RowData rowData : parsedRows) { String docId parseDocId(rowData); EventType eventType rowData.getEventType(); if (eventType EventType.DELETE) { bulkRequest.add(new DeleteRequest(product_index, docId)); } else { MapString, Object source buildSource(rowData); bulkRequest.add(new IndexRequest(product_index) .id(docId) .source(source)); } } BulkResponse response esClient.bulk(bulkRequest, RequestOptions.DEFAULT);IndexRequest天然具备幂等性同一个_id第二次写入就是整体覆盖。所以即使在回放过程中发生重复投递只要顺序不乱最终状态一定是对的。DELETE也是幂等的删不存在的文档不会报错。批量大小要控制好。单批太多会撑爆ESHTTP请求体太少又浪费吞吐。一般经验是攒到5000条或者1MB数据就flush一批两个条件先到先触发。这个阈值不是死的要看单条文档大小、ES节点数量和网络延迟。3.4 全量数据和增量数据的衔接Canal只能从启动那一刻开始订阅增量MySQL里已经存在的历史数据它不会自动搬到ES。所以一次完整的同步往往分为两步先做一次全量同步再启动Canal做增量同步。这里有一个经典的顺序做法。第一步在MySQL上执行SHOW MASTER STATUS;记录下当前的File和Position记为位点P。第二步用全量同步工具把当前MySQL数据导入ES。这里必须保证全量工具读取的是同一个时间点的一致性快照例如mysqldump默认就是一致快照。第三步配置Canal的初始位点为P启动增量消费。这个过程有一个很多人担心的点全量同步期间MySQL有新写入怎么办其实不会丢。Canal从P位点开始消费会把P之后发生的所有变更都回放一遍。即使全量ES已经包含了其中一部分变更回放时用_id覆盖写最终结果与MySQL一致。核心前提就是全量ES的初始快照和位点P要对齐否则会出现覆盖顺序问题。Canal的instance配置里可以通过这两个参数设置初始位点canal.instance.parser.binlogName mysql-bin.000001 canal.instance.parser.position 1234564. 生产环境中最容易踩的坑与排查方法跑demo和跑生产完全是两回事。下面这几个问题我在实际使用中几乎都遇到过有些甚至花了半天时间排查。4.1 典型问题速查表现象常见原因解决办法Canal启动后一直连不上MySQLMySQL账号权限不足或binlog未开启检查GRANT权限、log_bin和binlog_format日志报SlaveId冲突和其他从库或Canal实例重复修改canal.instance.mysql.slaveId能启动但收不到任何数据filter正则没写对用test_db\\.product格式注意点号转义收到数据但字段是乱码connectionCharset设置错误改成UTF-8并确认MySQL客户端编码大事务发生时消费端积压bufferMemory太小调大canal.instance.memory.bufferMemory消费端重启后数据重复上一批没ack或ack丢失改用getWithoutAck并保证幂等写入ES中出现幽灵文档删除事件没被处理检查DELETE事件构造DeleteRequestES写入报403/413bulk大小超过ES限制减少单批条数或体积4.2 数据一致性问题顺序、乱序、重复Canal单实例内部基本保证binlog顺序但下游如果并发消费乱序风险就会冒出来。举个例子一条UPDATE把价格改成100紧接着一条UPDATE改成80如果两个事件被并行线程处理先处理80再处理100ES最终价格就是100错了。所以并行消费时一定要按主键hash分流。最简单的做法是线程池内的任务队列按docId取模分配保证同一个主键的变更进入同一个队列串行执行。Canal带来的顺序性只能保证到“单实例顺序解析”能不能保持到ES写入取决于你自己。重复消费的问题没有银弹唯一靠谱的就是把ES写入设计成幂等的。任何场景下都用显式_id更新事件做整体覆盖而不是字段累加删除事件无所谓重不重复。4.3 消费端宕机后如何快速恢复TCP模式下Canal Server的位点保存在内存和meta文件里。只要消费端处理完一批调用了ackCanal就会记录这个位点。消费端宕机重启后从没有ack的最后一批开始重新拉取这就是“至少一次”语义。为了避免服务端内存丢了位点或者meta文件损坏很多团队会在消费端自己维护一份offset快照。每处理完一批把batchId和对应的binlog位点写到本地文件或数据库。重启时先从自己的存储恢复再用connector.rollback(batchId)回退到指定位置。如果不想写得这么复杂至少要在Canal配置里启用zookeeper模式让位点持久化到ZK。Canal Server本身也不建议单点部署。生产上可以部署两个Canal节点用canal.zkServers注册同一个ZK集群同一个instance的多个节点互为主备。客户端连接时通过ZK发现可用节点。这个高可用方案是Canal自带的比自己在消费端做重试要省心。5. 性能优化与几点实操体会当数据量从几十万涨到几千万增量同步的瓶颈往往不在Canal而在消费端和ES写入。5.1 分批和并行怎么平衡消费端拉取消息时connector.getWithoutAck(100)里的100只是单次最多拉取条目数不是硬性限制。实际返回条数取决于Canal缓冲区里积压了多少。积压很多时这一批可能达到几千条处理时间较长ES bulk的条数也会增多。如果单批过大ES bulk写入会超时这时可以限制单批处理数量比如攒够1000条就flush剩下的留着下批。具体实现是加入一个简单的计数器不等到所有消息解析完再写入。并行度建议从2到4开始按主键hash分配到不同队列。不要为了吞吐无脑调大线程数ES的写入瓶颈通常在CPU和磁盘IO线程过多反而会造成线程切换开销。5.2 Canal侧参数调优参考如果Canal解析速度跟不上先看canal.log里有没有full或者discard相关告警有的话说明缓冲区满了。调大bufferMemory虽然有效但同时也要确认Canal运行时的堆内存足够。比如bufferMemory调到1GCanal进程的-Xmx至少要大于2G否则GC频繁反而更慢。Canal解析binlog默认是单线程。如果是高并发写入的表可以适当调大canal.instance.parser.parallelThreadSize到4或8。但要注意这个参数影响的是Binlog解析不是消费投递。解析线程无论如何并行最终分发的顺序还是保持在队列里所以不用太担心乱序。5.3 我实际用下来觉得最值得注意的三件事第一消费端必须自己画清楚“Canal消息到ES文档对象”的映射函数不要直接在业务线程里写SQL查表组装宽表。每次增量都查一次MySQL高流量下会把主库连接池打爆。正确做法是在启动时加载一次关联映射缓存或者在CDC之前就通过视图把宽表在MySQL侧拼好。第二时间字段一定要统一时区。Canal解析出来的datetime和timestamp通常是字符串ES索引默认按UTC存储。如果消费端不做时区转换搜索时查询结果会整体差8小时。建议在消费端将所有时间字段统一成ISO格式并手动加上时区偏移。第三别把所有逻辑都塞进一个消费进程。Canal最常见的优势是同一份binlog可以同时服务多个下游ES负责搜索、Redis负责缓存、数仓负责分析。如果全部写在一个进程里任何一个下游异常都会拖垮整体消费进度。按instance拆分一个instance一个职责出问题只影响局部。同步管道这种东西配置写完只是起点真正值钱的是能持续稳定运行。整理这篇文章的时候我又把从MySQL到ES的每一步过了一遍越到后面越觉得核心不在于Canal本身而在于消费端的工程化程度。能把幂等、批量、位点恢复、乱序控制这几点想清楚这套管道就能长期扛住业务压力。