企业官网建设流程全解析

先说一下我这边为什么会折腾这个东西。之前接了一个业务项目，业务方要求把MySQL里的订单和商品数据同步到Elasticsearch（后面统一叫ES）里做搜索和聚合，实时性要求很高，数据变更后几秒钟内要能查到。最开始的方案很朴素，业务代码里双写，写MySQL的同时写ES，结果交互链路一长，两边经常对不上，还得靠定时任务每天全量对账，光维护就够头疼的。后面我换成了canal同步MySQL数据到ES这套方案，监听binlog增量变化，业务代码零侵入，MySQL主库写完，变更事件自己就会消费到ES里。这套方案我前后跑了小半年，从部署、全量初始化、增量订阅到各种坑基本都踩了一遍，这篇就把完整的实操过程、配置思路和排障心得整理出来，给同样想用canal同步MySQL数据到ES的读者一个可以直接抄作业的参考。

1. 为什么选canal这套方案：主流同步路径对比与取舍

先别急着上配置，同步方案选型这一步很关键，选错了后面返工成本极高。我梳理下来，MySQL数据同步到ES的主流路径大概有四条，每条的适用场景和坑都不同。

1.1 业务双写与定时任务方案的痛点

第一条路是业务双写：在业务代码里，凡是涉及订单表、商品表变更的地方，同时更新MySQL和ES。这种方式在小规模、表数量少、团队协作简单的时候能用，但痛点很明显。首先是事务问题，MySQL写成功了，ES写失败怎么办？回滚MySQL吗？很多场景根本没法回滚，只能靠补偿任务。其次是代码侵入，每张表、每个写接口都要改一遍，业务代码里到处是ES索引更新的逻辑，随着调用方增多，压根维护不住。最后是实时性依赖业务成功返回，一旦某个写操作失败被重试，ES里可能已经写了脏数据。

第二条路是定时全量同步：写个定时任务，每隔几分钟把MySQL数据全量捞一遍再刷进ES。这种方式适合数据量小、实时性要求不高的场景，但数据量一旦上了百万级，全量扫描对MySQL主库压力非常大，而且同步的频率上不去，无法做到秒级。早期我接手一个旧系统用的就是这种方案，订单表三百万行，每次全量同步耗时几分钟，ES里的数据最少延迟五分钟起步，搜索场景还能凑合，一旦业务方要求查实时库存和最新订单状态，这套方案基本就废了。

1.2 基于binlog的增量订阅为什么更合适

第三条路是引入消息队列做解耦：业务代码把变更事件发到消息队列，再由消费端写入ES。这个方案比双写干净一些，但还是在业务代码里埋了发消息的逻辑，而且消息队列本身也是额外的组件，运维成本不低。同时事件的可靠性依赖业务对消息的成功发送，一旦漏发，问题很难排查。

第四条路就是based on binlog的增量订阅方案，也就是canal干的活。canal的核心思路是把自己伪装成MySQL的从节点，向MySQL主库请求binlog，把数据库的插入、更新、删除事件解析出来，然后通过adapter或者自定义客户端消费到目标端。这个方案的好处是：

• 业务代码零侵入，不需要改任何业务逻辑；
• 数据链路天然跨过业务层，只要MySQL产生binlog就会触发同步，实时性可到秒级甚至毫秒级；
• 不额外引入消息中间件，canal组件本身够轻量（如果单机部署只需要一个Java进程加配置）。

1.3 canal同步MySQL到ES的整体链路

整体链路很清晰：MySQL主库开启binlog，canal server伪装成从库连接主库并解析binlog，canal client或canal adapter拿到解析后的数据变更事件，按配置的映射关系写入ES索引。

MySQL binlog → canal server（解析） → canal adapter（消费与映射） → Elasticsearch

这套链路我拆解成四个核心部分：MySQL端binlog配置、canal deployer部署、全量初始化方案、adapter映射配置。下面逐一展开。

2. 部署前必做的MySQL配置与canal环境准备

2.1 MySQL开启binlog，这一步不能省

canal工作的前提是MySQL已经开启binlog，且binlog格式必须设置为ROW。如果MySQL当前没开binlog，需要修改配置文件并重启MySQL实例，这一步通常需要和DBA确认，因为重启主库实例会影响线上连接。

MySQL端关键配置项：

[mysqld] # 开启binlog日志 log-bin=mysql-bin # binlog格式必须使用ROW，canal才能拿到每行数据变更前后的完整字段 binlog_format=ROW # 确保binlog中包含完整的前镜像和后镜像，删除操作才能拿到被删行的所有字段 binlog_row_image=FULL # 每个binlog文件大小，默认即可 max_binlog_size=128M # binlog保留天数，根据磁盘容量和数据恢复需求设置 expire_logs_days=7

需要注意，有部分云数据库默认不开启binlog，或者binlog_row_image不是FULL，需要工单申请开启。我踩过一次坑，MySQL开启了binlog但binlog_row_image是MINIMAL，删除数据事件里只带主键，不带其他字段，ES索引里其他字段的删除逻辑只能靠主键关联MySQL再查一次才能补全，非常麻烦。所以这一步务必确认是FULL。

权限方面，canal连接MySQL需要单独的账号。给它授权SELECT、REPLICATION SLAVE和REPLICATION CLIENT权限即可：

CREATE USER 'canal'@'%' IDENTIFIED BY 'canal_passwd'; GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%'; FLUSH PRIVILEGES;

这里SELECT权限是canal在特殊场景下回查数据用的，比如某些binlog事件信息不完整时需要通过主库补查，虽然实际同步主要靠binlog，但权限尽量给全，避免后续未知报错。

2.2 canal deployer安装与实例配置

canal分为deployer和adapter两部分。deployer负责连接MySQL和解析binlog，adapter负责把解析出来的数据写入目标端。这两部分可以部署在同一个节点，也可以分开部署。生产环境我建议至少用三台机器或者三个容器分别跑deployer的实例，配合ZooKeeper做高可用。我这里只讲单机部署链路，高可用集群在文章最后简单提一句。

部署时先从官方仓库下载对应版本安装包，解压后结构大致如下：

canal.deployer/ ├── bin/ # 启动停止脚本 ├── conf/ │ ├── canal.properties # canal server级别的全局配置 │ └── example/ │ └── instance.properties # 单个数据同步实例配置，example是默认实例名 └── lib/ # 依赖jar包

核心配置集中在canal.properties里的这几项：

# 当前节点角色，单机直接填1 canal.registerMode = tcp # server模式，固定就行 canal.serverMode = tcp # 实例列表，多个实例用逗号分隔 canal.destinations = example # 一批事件的最大字节数 canal.instance.memory.batch.mode = MEMSIZE

其实单机跑通的话，这些默认配置基本不用改，重点改的是conf/example/instance.properties。

实例配置里需要指定MySQL连接信息、监听过滤规则和binlog位点配置：

# MySQL主库地址，改成你自己的 canal.instance.master.address=127.0.0.1:3306 canal.instance.dbUsername=canal canal.instance.dbPassword=canal_passwd canal.instance.connectionCharset=UTF-8 # 要监听的库表，支持正则表达式 # 这里的规则表示监听test库下所有的表 canal.instance.filter.regex=test\\..* # 也可以指定多库多表，例如：test\\..*,shop\\.(order|goods) canal.instance.filter.black.regex= # 同步位点存储位置，单机用memory即可 canal.instance.global.mode=memory # 也可以指定为zookeeper，用于多节点高可用 # canal.instance.global.mode=zookeeper # canal.instance.zookeeper.host=127.0.0.1:2181

canal.instance.filter.regex这个配置我多说一句，第一次配置时很容易被反斜杠转义搞晕，test\\..*表示匹配test库的所有表。如果只想同步某一张表，写成test\\.order即可。正则匹配规则是canal基于Java正则的，写完之后建议先用grep确认配置文件里没有多余空格或者注释行被错误打开，否则实例启动会直接失败。

2.3 启动deployer与验证连接状态

配置完成后，先运行bin目录下的启动脚本：

./startup.sh

启动后去logs目录看日志，重点看两个文件：

• logs/canal/canal.log：server级别的日志，看是否完成启动
• logs/example/example.log：实例日志，看是否成功连上MySQL并dump binlog

实例启动成功后，日志里会出现类似dump address /127.0.0.1:3306 has been dump successfully的提示，这说明canal已经成功连接MySQL并开始拉取binlog。此时再到MySQL主库执行一条insert操作，实例日志会立刻出现对应表的DML事件解析信息。

需要注意，canal第一次启动默认是从MySQL当前最新的binlog位点开始消费的，也就是只能捕获启动之后的新增变更。如果ES里需要老数据，就得先做全量初始化，这一点下面专门讲。

3. 数据同步链路设计：全量初始化与增量订阅衔接

3.1 为什么必须先做全量初始化

canal本质上是增量订阅，它只同步启动之后产生的binlog事件，启动之前MySQL里已有的存量数据它是看不见的。所以整套同步链路的第一步，必须把存量数据先灌入ES，然后才能放心地让canal持续增量同步。

全量初始化有几种做法，我分别说下优劣。

方法一是直接用mysqldump导出数据，再写脚本批量写入ES。这种方式简单直接，但需要自己处理类型映射和批量写入逻辑，而且数据量大时对MySQL和ES都有压力。方法二是用canal自带的适配器，在adapter的配置里允许通过http接口触发全量同步，它会自己从MySQL拉数据并写入ES，比较省事。方法三是用中间表或时间字段分段捞，从一个update_time字段判断，把历史数据分批同步。这个适合不想停业务、且业务表有统一的修改时间字段的场景。

我实际推荐的做法是：先停掉canal的增量消费或者让增量消费处于暂停状态，完成存量全量导入，再把canal实例的消费位点对齐到全量导出的起点时间之后继续消费。这样能尽量避免数据重叠或漏数据。

具体来说，我曾经在某个项目里用过的简单方案是：

1. 记录一个时间点T1；
2. 用SELECT把业务表按主键分批捞出来，批量写入ES；
3. 等全量写完，再启动canal增量同步。

这里有个隐蔽问题：如果在T1到增量真正启动之间，业务表发生了新的变更，这一段的变更如果没被全量捞到，就会出现窗口期数据缺失。严谨一点的做法是，一边让canal的binlog消费位点保持在T1时刻，全量导入完成后，再让canal从T1继续消费，这样T1之后的新变更会以增量形式补偿进ES，不会丢数据。实际实现时，可以将canal的canal.instance.master.journal.name和canal.instance.master.position手动设置为T1时刻对应的binlog文件名和位点。

3.2 增量同步的批量消费与ack机制

增量同步启动后，canal client或adapter会持续拉取解析后的事件。canal在消费机制上采用的是批量拉取+确认ack的模型：客户端每次从canal server拉取一批事件，处理完成后发送ack确认；如果客户端处理失败，可以发送rollback，canal会从上次ack的位置重新推送这批数据。

这个机制保证了数据不丢，但也带来一个问题：如果没有正确实现ack逻辑，客户端进程一重启，可能会重复消费已经写入ES的数据。比如一批数据里有一部分写入ES成功，另一部分失败，如果整体回滚，之前成功写入的这部分又会重新消费一遍，如果不做幂等处理，ES里就会产生重复或脏数据。

在实际代码里，我简化的adapter客户端伪逻辑是这样：

while (running) { Message message = connector.getWithoutAck(batchSize); long batchId = message.getId(); List<CanalEntry.Entry> entries = message.getEntries(); if (batchId == -1 || entries.isEmpty()) { // 没有数据，稍等片刻 Thread.sleep(100); continue; } try { // 解析并写入ES，这一步必须保证整体成功 processEntries(entries); connector.ack(batchId); // 确认消费完成 } catch (Exception e) { logger.error("process batch error, rollback", e); connector.rollback(batchId); // 回滚，让canal重新推送 } }

这个逻辑看起来简单，但唯一需要谨慎的是processEntries这一层要保证整体成功。如果批次内部分事件写ES成功、部分失败，一旦回滚重推，成功的那部分也会再写一次，这时候ES端的写入必须做幂等。好在写入ES时可以用业务主键作为_id，同一主键重复写入在ES里本质是覆盖更新，所以重复消费的副作用基本可控。

3.3 表结构到ES索引映射：从MySQL表到index mapping

同步最核心的工作其实是表结构到ES索引映射的设计。MySQL的表是一对一的固定字段结构，ES的索引则可以支持嵌套、数组、keyword、text等多种类型，映射设计直接影响搜索效果和写入稳定性。

以一张商品表为例：

CREATE TABLE `goods` ( `id` bigint(20) NOT NULL AUTO_INCREMENT, `goods_name` varchar(128) DEFAULT NULL, `category_id` bigint(20) DEFAULT NULL, `price` decimal(10,2) DEFAULT NULL, `status` int(11) DEFAULT NULL, `create_time` datetime DEFAULT NULL, `update_time` datetime DEFAULT NULL, PRIMARY KEY (`id`) ) ENGINE=InnoDB;

对应ES的索引mapping，我一般遵循以下几条经验：

• 主键字段直接映射为ES文档的_id，保证写入幂等；
• 需要精确匹配的字段（状态值、分类ID）用keyword类型；需要分词搜索的字段（商品名）用text类型并指定中文分词器；
• 金额字段最好用scaled_float或double，避免使用text导致无法范围查询；
• 时间字段统一映射为date类型，并严格按照yyyy-MM-dd HH:mm:ss或strict_date_optional_time格式写入，否则ES解析时间字段会报错。

这部分映射关系在canal adapter里可以通过配置完成，也可以在adapter的client代码里自己控制。如果用的是canal adapter的es模式，配置文件里会有一份索引映射的规则，我在下一节详细讲。

4. 适配器接入：canal adapter同步到ES的配置实操

4.1 配置adapter的数据源与ES连接

canal官方提供adapter组件，它内置了多种目标端的写入能力，其中就包括Elasticsearch。adapter启动后会订阅canal server的实例，通过配置的映射规则把数据自动写入ES，省去自己写客户端的功夫。

先解压canal.adapter，修改application.yml里关于数据源和ES连接的配置：

canal: conf: # 对应deployer里配置的实例名 canalAdapters: - instance: example groups: - groupId: g1 outerAdapters: - name: logger - name: es7 hosts: http://127.0.0.1:9200 properties: cluster.name: my-elasticsearch # 连接ES时用的账号密码，如果ES开启了安全认证 security.auth: elastic:elastic_passwd

这里s7是ES 7.x版本的适配器key，如果你用的是ES 8.x需要确认对应适配器的兼容性。早期我用ES 7.4搭配adapter，整体比较稳定，切换到ES 8.x后适配器版本不匹配，REST API的兼容层有些字段变了，花了不少时间排查。建议先确认adapter版本对ES大版本的兼容支持再决定。

4.2 映射规则文件的核心配置

adapter在conf目录下会有一个映射目录，每个实例对应一个配置文件，比如order.yml或goods.yml。以同步goods表到goods_index索引为例：

dataSourceKey: defaultDS destination: example groupId: g1 esMapping: _index: goods_index _type: _doc _id: id sql: "SELECT id, goods_name, category_id, price, status, create_time, update_time FROM goods" commitBatch: 3000 etlCondition: "where update_time >= {0}" mapping: id: id goods_name: goodsName category_id: categoryId price: price status: status create_time: createTime update_time: updateTime

逐项拆解：

• _index：目标ES索引名；
• _id：指定文档ID对应的MySQL字段名，这里用主键id，能保证同一个主键的变更在ES里覆盖更新而不是新增；
• sql：adapter执行ETL和全量初始化时用的查询SQL，字段顺序会作为后续映射的上下文字段；
• mapping：MySQL字段到ES字段名的映射关系。注意canal adapter的es同步模式里，mapping的左侧是SQL查询结果里的字段名，右侧是ES索引里的字段名，如果两边一致也要写出来；
• commitBatch：批量提交的条数，根据数据量调整，太高会导致ES写入压力大，太低吞吐又起不来。

这里有一个很关键的点：adapter的es同步逻辑其实是先通过SQL查询出当前主键对应的整行数据，再整体覆盖写入ES。这就意味着，不管binlog事件是插入、更新还是删除，adapter的处理方式是：如果是删除事件，根据_id执行delete请求；如果是插入或更新事件，则执行SQL查询该主键的最新整行数据，再整体写入ES。所以配置里的SQL字段必须覆盖ES索引里需要的所有字段。

4.3 同步细节：更新、删除与字段变更的处理

插入和更新事件adapter处理方式一样：拿binlog事件里的主键ID，按照配置的SQL重新查MySQL，把最新整行数据取出来写入ES。这个设计的好处是即使binlog的ROW事件里只包含变更字段，ES里最终写入的也是完整的一行数据，不会因为字段缺失导致索引字段为空。

删除事件则走ES的delete接口。但这里有个容易忽略的问题：如果删除业务表的记录时，MySQL的binlog事件里主键字段必须存在。使用FULL格式的binlog image可以确保删除事件里携带全字段数据，但adapter其实只需要主键。之前遇到过一种情况，业务方在某个历史版本中直接删除了数据库记录，没有通过应用层删除，canal推送的删除事件主键是对的，但ES里对应文档删除后，关联的二级缓存没清干净，搜索端还残留旧数据。这块需要业务侧自行处理缓存失效逻辑。

另外映射时要特别注意MySQL字段类型与ES类型的对应。比如MySQL的decimal(10,2)在ES里如果用keyword接收，写入时会变成字符串，后续如果要范围过滤、按金额排序，搜索就会出问题。我一般的做法是：MySQL的数值类型字段在SQL查询里用CAST转成ES期望的数值类型，例如CAST(price AS DECIMAL(10,2))，或者让映射字段在ES里定义为scaled_float。简而言之，binlog解析出来的是MySQL的原始类型，最终写入ES的JSON格式取决于adapter的转换逻辑和映射定义，建议在正式接入前用几条真实数据做一次端到端验证。

5. 生产环境实战：性能调优与高频踩坑现场

5.1 写入延迟从秒级到毫秒级的调优

canal同步MySQL数据到ES，数据从MySQL binlog产生到ES可搜索，中间经过canal解析和adapter处理。默认配置下，端到端延迟通常在一秒左右，这是符合预期的，毕竟canal是批量消费机制，不会每条事件都单独写一次ES。

如果发现延迟超过三秒甚至分钟级，优先排查以下几点：

• canal server到MySQL主库的读取速率：查看example.log里是否有blocking字样的日志，如果出现，说明deployer解析binlog的速度跟不上binlog的产生速度，简单粗暴的解决办法是调大instance的批量消费字节数：

  canal.instance.memory.batch.size=4096 cannal.instance.memory.buffer.size=16384

  同时检查deployer所在机器的磁盘IO和网络，binlog解析是连续读盘操作，磁盘IO高会直接影响解析速度。

• adapter批量提交设置：commitBatch设置的太小（比如100），数据量大时会导致ES写入太频繁，反而拖慢整体写入性能。我这里的经验值：中等数据量（单表日增百万行以内）设3000到5000比较合适。

• ES端bulk写入队列：adapter写入ES走的是bulk接口，如果ES出现写入拒绝或circuit_breaker报错，说明ES节点写入压力太大。此时去检查ES集群的堆内存和磁盘阈值，及时扩容或调整bulk并发参数。

延迟调优最直观的验证方式：在MySQL里做一条update，在ES端通过文档的_seq_no或者更新时间字段观测到可见时延。如果压测后发现瓶颈不在canal而在ES端，不要盲目调大canal的batch，先把ES的写入吞吐提上来才是正解。

5.2 字段映射不当引发的写入失败

这是一类我在接入过程中遇到最多的问题，基本集中在类型不匹配上。

比如MySQL的datetime字段，adapter在生成JSON时会输出类似2024-06-01 12:00:00的字符串，ES mapping里如果对应字段是date类型且没有指定format，ES会自动尝试解析这种格式，一般是能成功的。但如果ES mapping里字段被定义成了keyword，写入也会成功，但后续你想按时间范围搜索就傻眼了，keyword根本没法做ranger查询。

更常见的是字段长度超出ES keyword限制。ES 7.x默认keyword类型最大长度是32766字节，MySQL里如果有一个varchar(50000)字段，映射成keyword后超长文本写入会直接报错。解决办法是这类长文本在ES mapping中定义为text类型，开启分词即可。

还有一个不显眼但容易踩的坑：字段名大小写不一致。MySQL字段默认是不区分大小写的，但是ES映射的字段名是区分大小写的。如果SQL查询里写的是ID，mapping右侧写的是id，adapter在做字段映射时会找不到对应字段，导致ES里该文档的部分字段为空或者整条写入失败。这个需要在配置完映射规则后用测试数据立刻验证，不能等上线后才发现。

5.3 幂等性与顺序性保障

前面提到canal的消费模型是批量拉取加ack确认，这会导致重复消费。尽管我们将业务主键映射为ES文档_id可以覆盖更新，但如果binlog事件本身携带的字段是旧值，重复消费时会用旧值覆盖新值。这种问题通常发生在adapter处理慢、消费位点回退、或者网络抖动导致ack丢失的场景。

举个例子：同一行数据先执行更新A，把status设置为2，再执行更新B，把status设置为3。如果canal推送了A和B两个事件，adapter在处理A时执行SQL查询，结果查到的是最新的status=3，随后的B事件也会查到status=3，两次写入ES都是status=3，没问题。但如果A事件处理极慢，消耗了很长时间，等A的SQL终于查到MySQL时，B已经又改成4了，A先把status=4写入，紧接着B事件再查询一次写入status=4，最终结果也对。 因为每次都是拿最新全量行覆盖写入，所以顺序问题在这个模式下被弱化了。

真正的顺序性隐患出现在删除事件上。如果一个主键在短时间内经历了更新和删除，adapter处理删除时，MySQL里已经查不到这一行了，它会执行ES delete。但如果删除事件处理失败被回滚，重推后MySQL里仍然查不到该行，ES里的delete会再次执行，通常会得到not_found响应，不影响最终结果，但会有大量重复日志。

如果这些事件走的是自定义客户端，而不是adapter模式，则必须在客户端逻辑中自行保证同一主键事件的顺序消费。最简单的方式是按主键哈希路由到单线程处理，避免同一行的多个binlog事件被并发消费。

注意：canal默认推送事件是按binlog顺序排列的，但在多线程消费时，如果事件被拆分处理，顺序可能被打乱。一定要避免同一条主键的记录被多个线程同时更新到ES，否则会出现最终值不一致。最稳妥的方案是单线程消费，或者在客户端内对主键做哈希后按分区串行处理。

5.4 DDL变更对同步链路的影响

线上业务调整表结构时会执行ALTER TABLE，这类DDL事件canal也能捕获，但adapter的默认处理往往不够完善。

我遇到过一个场景：业务方给商品表新增了一个字段，执行完ALTER TABLE之后，adapter日志开始报字段找不到的错误，原因是映射配置里的SQL还是旧的字段列表，adapter执行SELECT时查询的字段已经不存在了，导致所有变更事件都失败。

这要分两步解决：一是调整adapter映射SQL，把新字段补上；二是在ES索引的mapping里增加对应字段（如果索引定义比较严格的话）。因此，DDL变更必须建立变更流程，不能直接在生产库上执行alter。如果有多个下游依赖，需要先在测试环境验证DDL对canal链路的影响，再在业务低峰期执行。

6. 常见问题与排障速查表

下面把我在实际运维中积累的典型故障和排查方法整理成一个速查表，每一条都对应真实场景，方便读者在出问题时快速定位。

在做排障时有一个习惯值得养成：所有同步问题先看日志，再看数据，最后看配置。canal的日志其实写得很清楚，尤其example.log里会打印每一条解析后的entry类型和表名，adapter日志会打印写入ES的实际请求和响应。多数问题通过日志就能定位，不需要乱猜。

还有一个排查技巧：canal自带的管理接口可以从canal.properties里配置的端口访问，通过管理接口能看到当前实例的消费位点、延迟、客户端连接数等信息。结合MySQL主库的SHOW MASTER STATUS里的binlog位点，可以快速判断canal是不是已经积压了大量未消费的binlog。如果二者位点差距持续拉大，说明消费链路某处出了瓶颈，直接查看adapter写入ES的耗时和错误日志即可。

7. 初始化快照生产实践：数据一致性验证与回滚预案

全量同步和增量同步衔接时，最大的风险是数据不一致。我只讲我自己用过的稳妥流程。

在正式对线上库做全量初始化前，先在一个临时表或者Shadow环境做一次小规模试验，这能帮你提早发现映射配置、SQL语句、ES mapping的问题。试验完成后，再在正式环境按下面步骤操作：

1. 记录启动时间点T0；
2. 把canal实例的消费位点设置到T0对应的binlog位置，让canal暂时不消费，只等待；
3. 配置好adapter的映射规则，手动触发adapter的全量ETL功能，它会通过配置的SQL把存量数据写入ES；
4. 全量ETL完成后，立即启动canal增量消费，因为它的位点在T0，T0之后发生的变更都会被补推；
5. 写一个比对脚本，从MySQL按主键随机抽1000条数据和ES对比，检查字段全量一致。

这里最容易出的问题是第4步和第5步之间可能有窗口期。如果全量ETL本身耗时很长，比如几千万行数据导了几个小时，那T0到ETL结束之间的新变更如果已经通过canal补推到ES，而ETL又把这些数据重复写了一次，就会出现短暂的数据回退。稳妥的做法是：全量ETL期间暂停canal增量消费，等ETL全部结束再启动，让增量从T0开始追平。

对于数据一致性校验，不要只比对主键是否存在，还要比对关键业务字段。因为ES中常见的问题不是数据不存在，而是部分字段因为映射问题为空或类型错误。校验脚本要覆盖数值、时间、文本几个类型，字段数量不需要太多，但每个字段类型的代表性必须有。

一旦在校验中发现某张表数据不一致，回滚预案是：直接从MySQL中按主键分批重新拉取该表全部数据，用adapter或脚本强制覆盖写入ES，覆盖完成后再重新启动增量消费。如果没有完整的覆盖写入能力，最坏的情况就是全量重导，所以初始化完成后，保留一份ETL用的SQL配置很重要，后续补数、对账都能复用。

8. 从单机到多活：canal高可用与规模扩展的一点经验

最后顺便说一句生产化的问题。单机canal在数据量小、没有高可用要求时完全够用。但一旦业务规模上来，比如Store主库每秒写入几千条记录，ES索引几十个，单机deployer往往撑不住，需要做高可用部署。

canal的高可用依托ZooKeeper，多个deployer节点组成集群，同一个实例（比如example）只能在其中一个节点运行，其他节点作为standby，当主节点宕机后自动切换。这是官方支持的模式，配置上需要在canal.properties打开：

canal.zkServer=127.0.0.1:2181 canal.instance.global.mode=zookeeper

同时，多实例的横向扩展也有两种路径。一种是按业务库拆分成多个canal实例，每个实例负责一组表，同一台机器可以跑多个instance，各自独立消费。另一种是同一个实例的消费端使用多个client，通过groupId实现客户端集群消费，类似于消息队列的消费组。

我用过的组合是：第一台机器跑4个deployer实例实例，分别对应A库、B库、C库、D库，adapter配置多个group或分组消费。这样的好处是，某一张表数据量暴涨导致同步延迟时，只影响对应实例，不会拖垮所有同步链路。

高可用方案会带来额外的运维成本，尤其是ZooKeeper和集群状态监控需要额外部署。如果你的团队运维能力有限，建议先把单机方案跑稳，把监控报警补齐，再逐步演进。很多时候，问题的瓶颈不在canal本身，而在数据模型和ES索引设计，这些基础的东西不夯实，上再多机器也白搭。

这套方案跑下来，最大的感受是：canal同步MySQL数据到ES，真正决定成败的往往不是canal本身的搭建，而是数据映射和链路容错。配置好binlog、设计好索引mapping、做好全量增量衔接、想清楚重复消费下的幂等，这四个点打通了，整个同步链路就很顺。我在实际使用中发现，adapter的SQL + mapping模式虽然灵活，但遇到复杂宽表和多表关联时还是得自己写消费段逻辑，不要硬套官方adapter。如果你刚开始接触，建议先拿一张简单表把端到端链路跑通，确认日志里事件解析、ES写入都正常，再逐步扩展到核心业务表。同步数据没有一劳永逸的方案，保持对监控日志的敏感度，才是长期稳定运行的关键。