企业官网建设流程全解析

我最早接触 MySQL 与 Elasticsearch 的数据同步时，还是靠写定时任务去扫表，把变更的数据查出来再灌进 ES。那会儿数据量小，感觉还能忍，但等数据一上来、字段一多、更新一频繁，整个方案就越来越别扭：延迟不可控、全表扫描压力大、业务代码也被同步逻辑搞得稀碎。后来把 Canal 引入到这套流程里，靠它订阅 MySQL 的 binlog，把增量变更实时解析出来，再同步到 ES，整个过程就轻松顺滑了很多。

Canal 是阿里巴巴开源的一个组件，核心思路是伪装成 MySQL 的从库，拉取主库 binlog，然后以 JSON 之类的格式把每一行数据的变更事件推送给消费端。你的“消费端”可以理解为一个独立的同步程序，自己决定要把事件写到 ES、Redis、MQ，还是其他任意存储。这篇文章我会从原理到实战，把 Canal 同步 MySQL 增量数据到 ES 这条链路完整拆解一遍，包括环境准备、配置项、消费代码怎么写、binlog 数据长什么样，以及我踩过的那些坑。适合正在搭数据同步管道、或者被“MySQL 改完数据后 ES 里对不上”折磨的同学参考。

1. 整体设计思路：为什么是 Canal 而不是双写或定时任务

1.1 常见同步方案的瓶颈

把 MySQL 数据同步到 ES，很多人第一个想法就是在业务代码里双写：写 MySQL 的同时再写一份 ES。这个方案对上线的业务侵入非常大，每一个涉及数据变更的地方都要改，一旦 ES 抖动还会反向拖垮主流程。更麻烦的是，历史数据根本没法通过双写解决，MySQL 里已经存在的几百万条旧数据还是得专门写脚本导一次。

定时任务扫表是另一个常见做法，实践过的人都懂痛点：要么只能增量同步“更新时间”字段，但很多表压根没有这个字段；要么每次全量轮询，对数据库压力大；要么同步窗口固定，实时性很差。我见过一个极端案例，某系统 5 分钟跑一次同步脚本，结果用户改完昵称后，在搜索结果里看到的还是老昵称，体验确实很差。

1.2 Canal 如何解决这些问题

Canal 的思路很聪明，它不主动去查业务表，而是把自己伪装成 MySQL 的 slave 节点。因为 MySQL 主从复制流程里，从库本来就是靠读取主库的 binlog 来同步数据变更的，Canal 做的事情和真实从库几乎一样：向 MySQL 注册成一个 slave，MySQL 就会持续把 binlog 推送给它。

这样带来的好处非常明显：

• 业务零侵入，代码不需要改任何一行，数据库层面能拿到完整的变更记录。
• 实时性很高，binlog 是 MySQL 产生事务日志时实时落盘的，Canal 拿到日志后马上就能解析推送。
• 可靠性有保障，binlog 是 MySQL 原生的复制能力，不会丢数据，Canal 自身还有位点（position）记录，崩溃后可以从上次的位置继续消费。

1.3 整体数据链路

从架构上看，这条链路非常清晰：

MySQL binlog --> Canal Server --> Canal Client（自研消费程序） --> Elasticsearch

Canal Server 负责跟 MySQL 打交道，解析 binlog；Canal Client 是嵌在你的应用程序里的一段代码，负责接收解析后的变更事件，然后调用 ES 的 API 把数据写进去。

我实际项目里遇到过一个情况：业务方那边不让部署额外的 Java 服务，只允许提供数据库账号。这种情况下就可以让 Canal Server 把数据投递到 Kafka，下游再由一个独立的 Go 或 Python 服务去消费。也就是说，Canal 本身并不强迫你一定要用什么语言或者什么架构，它可以把解析后的数据推给 TCP 客户端、Kafka、RocketMQ 或者 Pulsar，灵活性很强。

2. 前置准备：MySQL 开启 binlog 与 Canal 环境部署

2.1 确认 binlog 开启状态

在正式折腾 Canal 之前，先得确认 MySQL 有没有开启 binlog，而且必须保证 binlog 格式是 ROW。Canal 只有在 ROW 格式下才能拿到某一行数据变更前后的完整字段值，如果用的是 STATEMENT 格式，binlog 里记录的是 SQL 语句本身，Canal 无法还原出数据行。

用下面的命令检查：

SHOW VARIABLES LIKE 'log_bin'; SHOW VARIABLES LIKE 'binlog_format';

如果 log_bin 是 OFF，或者 binlog_format 不是 ROW，需要修改 MySQL 配置文件。以 MySQL 5.7 / 8.0 为例，在my.cnf或my.ini的[mysqld]段里加上：

server-id=1 log_bin=mysql-bin binlog_format=ROW binlog_row_image=FULL expire_logs_days=7

这里有个比较坑的细节：server-id一定要设置，而且不能跟现有复制拓扑里的其他节点重复。Canal 伪装从库时也需要一个 server-id，如果两边的 server-id 冲突，MySQL 会直接断开复制连接。

改完配置后重启 MySQL，再用上面的命令确认。binlog 开启后 MySQL 性能会有轻微损耗，但绝大多数业务场景完全可以接受。

2.2 创建 Canal 要用到的 MySQL 账号

Canal 读取 binlog 不是用普通 SELECT 权限，它需要REPLICATION SLAVE、REPLICATION CLIENT这两个复制相关的权限。给 Canal 单独建一个账号是常规做法，不要直接拿 root 去跑：

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

如果是 MySQL 8.0 版本，还需要注意默认认证插件的问题。MySQL 8.0 默认用caching_sha2_password，Canal 1.1.4 之前的版本对此支持不好。我一般会显式指定成mysql_native_password：

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

2.3 部署 Canal Server

Canal Server 的部署方式有几种：直接下载发行包、用 Docker 跑，或者用 Kubernetes 部署。我个人的习惯是在本地测试时用 Docker，在正式环境用发行包跑在独立的机器上，这样做的好处是资源和版本都隔离得干净。

用 Docker 部署 Canal Server 比较直接：

docker run --name canal-server \ -p 11111:11111 \ -e canal.instance.master.address=mysql-host:3306 \ -e canal.instance.dbUsername=canal \ -e canal.instance.dbPassword=canal_password \ -e canal.instance.connectionCharset=UTF-8 \ -e canal.instance.tsdb.enable=true \ -d canal/canal-server:v1.1.6

不过要说明一下：我踩过一次坑，用 Docker 部署时如果 MySQL 和 Canal 不在同一个 Docker 网络里，master.address千万别写localhost或127.0.0.1，必须写宿主机在容器网络里能访问的 IP，否则 Canal 一直报连接不上。

如果选择发行包方式，步骤大概是：

1. 从 GitHub Releases 页面下载 canal.deployer 的 tar.gz 包。
2. 解压后进入conf/example目录，编辑instance.properties。
3. 启动bin/startup.sh，日志在logs/example/example.log。

2.4 Canal Adapter 与 Canal Client 怎么选

在配置之前，还有一个方案需要先想清楚：Canal 生态里同步数据到 ES 其实有两种路线，一种是使用官方的 Canal Adapter，另一种是自己写 Canal Client。

Canal Adapter 是官方推出的开箱即用同步工具，它能通过配置规则文件，直接把 MySQL 表映射到 ES 索引，不需要写代码。优点是省事，缺点是灵活性一般，复杂映射、多表关联、字段加工做起来比较别扭。

我的建议是：如果只是简单的单表字段映射，没有太多业务逻辑，直接用 Adapter 最快；如果你需要对数据做加工，比如把几个字段拼接成一个 ES 字段、做类型转换、或者接到消息队列再处理，那就自己写 Client。这篇博文后面主要讲 Client 的方式，因为这是绝大多数人的真实需求，也是理解 Canal 工作原理最好的路径。

3. 核心细节解析：Canal 配置项与 binlog 数据格式

3.1 关键配置项说明

Canal 的配置主要集中在两个文件：canal.properties和instance.properties。前者是 Server 级别的全局配置，后者是单个数据通道的配置。

canal.properties里最重要的几个配置：

# 当前 Server 注册到 zookeeper 的标识（集群模式才用） canal.zkServers= # tcp, kafka, rocketmq, pulsar 等模式 canal.serverMode=tcp # 端口 canal.port=11111 # 每个实例占用的内存 canal.instance.memory.buffer.size=16384

instance.properties是单通道配置，里面最关键的就是 MySQL 连接信息和 binlog 位点：

# MySQL 地址和账号 canal.instance.master.address=mysql-host:3306 canal.instance.dbUsername=canal canal.instance.dbPassword=canal_password canal.instance.connectionCharset=UTF-8 # binlog 位点，第一次启动时可以不填，Canal 会自动从最新的 binlog 位置开始 canal.instance.master.journal.name= canal.instance.master.position= canal.instance.master.timestamp= # 需要订阅的库表规则，支持正则 canal.instance.filter.regex=test_db\\..* # 黑名单 canal.instance.filter.black.regex=

这里重点说一下canal.instance.filter.regex。默认配置是.*\\..*，表示订阅所有库的所有表。实际生产环境我建议收敛到指定的库和表，既能减少无用事件，也能避免把一些敏感表的数据也同步出去。规则写法是库名\\..*，其中库名和表名中间是两个反斜杠加一个点。

位点的含义值得展开：Canal 通过master.journal.name（binlog 文件名）和master.position（文件内偏移量）来定位自己该从哪里开始读取。第一次启动时，如果这两个配置留空，Canal 会默认从当前最新的 binlog 位置开始；如果你希望它从某个历史时间点开始解析，就可以手动指定位点。但要注意，binlog 文件是有过期时间设置的（expire_logs_days），指定太早的位点而 binlog 已经清理掉的话，Canal 会报错找不到文件。

3.2 binlog 解析后的数据长什么样

理解 Canal 输出的数据格式，是写好消费端代码的前提。Canal 的每条变更记录本质是一个事件，里面包含表名、库名、事件类型（增删改）、变更前数据、变更后数据。

下面是一个 INSERT 事件的典型 JSON：

{ "id": 12345, "database": "test_db", "table": "user", "es": 1700000000000, "isDdl": false, "type": "INSERT", "ts": 1700000001000, "data": [ { "id": "1001", "name": "张三", "age": 25, "status": 1 } ], "old": null }

字段含义：

• type：事件类型，INSERT/UPDATE/DELETE。
• database和table：变更发生的库和表。
• data：变更后的当前行数据，是一个数组，通常只有一个元素；如果涉及批量操作可能会有多个。
• old：变更前的旧值，只有 UPDATE 事件里才会有，而且只包含被修改的字段。
• ts：事件时间戳。
• es：执行时间戳，MySQL 端执行事务的时间。

看一个 UPDATE 事件的例子就更能理解old的用途了：

{ "id": 12346, "database": "test_db", "table": "user", "type": "UPDATE", "data": [ { "id": "1001", "name": "李四", "age": 26, "status": 1 } ], "old": [ { "name": "张三", "age": 25 } ] }

消费端拿到这种事件后，直接用data[0]覆盖写入 ES 文档即可。old里的数据一般用于日志审计，或者只在特定场景下使用（比如判断某个字段是否被修改过再决定是否重新同步关联数据）。

DELETE 事件只有data，没有old。消费端收到后要做的事情就是从 ES 里删除对应 id 的文档。

3.3 同步到 ES 的索引设计

在动手写消费代码之前，索引设计是绕不开的一环。Canal 消费端写 ES 时，通常直接用 MySQL 的主键作为 ES 文档的_id，这是最稳妥的做法。

原因很简单：能保证同一个业务实体在 ES 里只有一份文档，重复消费同一事件也只会覆盖同一份文档，天然幂等。启动同步任务时如果用自增 ID 作为 ES_id，一旦同步任务重跑，ES 里就会出现重复的文档。

索引 mapping 设计上，需要注意 MySQL 字段类型和 ES 字段类型的对应关系。比如 MySQL 的bigint在 ES 里如果映射成long没问题，但unsigned bigint超出long范围就可能出问题；MySQL 的datetime建议在 ES 里用date类型存储，并明确指定格式，避免默认格式解析失败。

4. 实操过程：从部署 Canal 到编写消费端代码

4.1 准备测试环境

为了把整个流程走通，我建议先在一台测试机上跑通最小链路。假设环境如下：

• MySQL：8.0.x，已开启 binlog，binlog_format=ROW。
• ES：7.x，已启动并创建好索引。
• Canal：1.1.6，运行在一台独立机器上。
• 消费端：一个简单的 Java 项目，用 Canal Client 接收事件并写入 ES。

测试库表结构：

CREATE DATABASE test_db DEFAULT CHARACTER SET utf8mb4; USE test_db; CREATE TABLE `user` ( `id` bigint(20) NOT NULL AUTO_INCREMENT, `name` varchar(64) NOT NULL, `age` int(11) DEFAULT NULL, `created_at` datetime DEFAULT CURRENT_TIMESTAMP, PRIMARY KEY (`id`) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

ES 索引定义可以先用一个简化的 mapping：

{ "mappings": { "properties": { "id": { "type": "long" }, "name": { "type": "keyword" }, "age": { "type": "integer" }, "created_at": { "type": "date", "format": "yyyy-MM-dd HH:mm:ss" } } } }

4.2 配置 Canal Server

假设 Canal 用发行包方式部署在/opt/canal目录。编辑conf/example/instance.properties：

canal.instance.master.address=127.0.0.1:3306 canal.instance.dbUsername=canal canal.instance.dbPassword=canal_password canal.instance.connectionCharset=UTF-8 canal.instance.filter.regex=test_db.user

然后启动：

/opt/canal/bin/startup.sh

查看启动日志确认成功：

tail -f /opt/canal/logs/example/example.log

如果日志里没有异常，并且出现类似“start successful”的信息，说明 Canal 已经成功连上 MySQL 并开始监听 binlog 了。

4.3 连接 Canal 并消费事件

Canal Client 是 Java 生态里的 SDK，接入方式很成熟。在项目的pom.xml中引入依赖：

<dependency> <groupId>com.alibaba.otter</groupId> <artifactId>canal.client</artifactId> <version>1.1.6</version> </dependency>

再引入 ES 的 Java High Level REST Client（ES 7.x）：

<dependency> <groupId>org.elasticsearch.client</groupId> <artifactId>elasticsearch-rest-high-level-client</artifactId> <version>7.17.x</version> </dependency>

然后写一个最基础的消费端：

package demo.canal; import com.alibaba.otter.canal.client.CanalConnector; import com.alibaba.otter.canal.client.CanalConnectors; import com.alibaba.otter.canal.protocol.Message; import com.alibaba.otter.canal.protocol.CanalEntry; import com.alibaba.otter.canal.protocol.CanalEntry.Entry; import com.alibaba.otter.canal.protocol.CanalEntry.EntryType; import com.alibaba.otter.canal.protocol.CanalEntry.RowChange; import com.alibaba.otter.canal.protocol.CanalEntry.RowData; import com.alibaba.otter.canal.protocol.CanalEntry.EventType; import java.net.InetSocketAddress; import java.util.List; public class CanalToEsWorker { private static final String CANAL_SERVER = "127.0.0.1"; private static final int CANAL_PORT = 11111; private static final String DESTINATION = "example"; private static final String USERNAME = ""; private static final String PASSWORD = ""; public static void main(String[] args) { CanalConnector connector = CanalConnectors.newSingleConnector( new InetSocketAddress(CANAL_SERVER, CANAL_PORT), DESTINATION, USERNAME, PASSWORD); int batchSize = 1000; try { connector.connect(); connector.subscribe("test_db\\..*"); connector.rollback(); while (true) { Message message = connector.getWithoutAck(batchSize); long batchId = message.getId(); if (batchId == -1 || message.getEntries().isEmpty()) { try { Thread.sleep(200); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } continue; } for (Entry entry : message.getEntries()) { if (entry.getEntryType() != EntryType.ROWDATA) { continue; } RowChange rowChange; try { rowChange = RowChange.parseFrom(entry.getStoreValue()); } catch (Exception e) { throw new RuntimeException("解析binlog事件失败", e); } EventType eventType = rowChange.getEventType(); for (RowData rowData : rowChange.getRowDatasList()) { switch (eventType) { case INSERT: handleInsert(entry, rowData); break; case UPDATE: handleUpdate(entry, rowData); break; case DELETE: handleDelete(entry, rowData); break; default: // 忽略DDL等事件 break; } } } connector.ack(batchId); } } finally { connector.disconnect(); } } private static void handleInsert(Entry entry, RowData rowData) { String database = entry.getHeader().getSchemaName(); String table = entry.getHeader().getTableName(); Map<String, Object> data = toMap(rowData.getAfterColumnsList()); // 调用ES API写入文档，_id使用主键 System.out.println("INSERT: " + database + "." + table + " -> " + data); } private static void handleUpdate(Entry entry, RowData rowData) { String database = entry.getHeader().getSchemaName(); String table = entry.getHeader().getTableName(); Map<String, Object> data = toMap(rowData.getAfterColumnsList()); Map<String, Object> old = toMap(rowData.getBeforeColumnsList()); // 调用ES API更新文档 System.out.println("UPDATE: " + database + "." + table + " -> " + data + ", old: " + old); } private static void handleDelete(Entry entry, RowData rowData) { String database = entry.getHeader().getSchemaName(); String table = entry.getHeader().getTableName(); Map<String, Object> data = toMap(rowData.getBeforeColumnsList()); // 调用ES API删除文档，_id用主键 System.out.println("DELETE: " + database + "." + table + " -> " + data); } private static Map<String, Object> toMap(List<CanalEntry.Column> columns) { Map<String, Object> map = new HashMap<>(); for (CanalEntry.Column column : columns) { if (!column.getIsNull()) { map.put(column.getName(), column.getValue()); } else { map.put(column.getName(), null); } } return map; } }

这段代码的运行逻辑是：拉取一批事件，遍历每一条 binlog 变更，根据事件类型分别处理，处理完成后调用ack确认消费成功。getWithoutAck这个 API 值得理解一下：它表示“取一批数据但先不确认”，只有当你处理完并调用ack后，Canal 才会认为这批数据消费成功了；如果中途程序崩溃，这批数据会在下次连接时被重新推送。这就是 Canal 提供的最基础的“至少一次”语义。

4.4 写 ES 的细节处理

上面代码里我用System.out.println代替了真实的 ES 写入，实际项目中需要把toMap的结果转成 ES 的IndexRequest或者UpdateRequest。这里有几个容易踩的细节：

第一，主键字段的处理。推荐用 MySQL 主键作为 ES 文档_id，这样天然幂等。如果业务表有复合主键，那就需要把多个字段拼接成一个字符串作为_id，拼接规则要稳定，避免同一条数据在不同事件里生成不同_id导致文档重复。

第二，null 值的处理。toMap里已经判断了isNull，null 字段在写入 ES 时如果 mapping 不允许 null，可能导致整个文档写入失败。我的建议是在消费端统一做一次过滤，把值为 null 的字段剔除掉再写入，反正 ES 中不存在的字段和 null 字段在查询行为上基本等价。

第三，时间字段格式。MySQL 的datetime返回的是一个字符串，比如2024-11-16 10:30:00。ES 的date类型默认支持的是 ISO 格式，比如2024-11-16T10:30:00.000Z。如果不做格式化，ES 解析会直接报错。我通常会在toMap里做一层转换，把datetime字段统一格式化成 ES 可以解析的格式，或者在 mapping 里指定"format": "yyyy-MM-dd HH:mm:ss"，两边对齐。

4.5 全量同步怎么配合

Canal 只解决增量同步的问题，但大多数项目在第一次接入时，ES 里其实是空的，或者只有一份过期的全量数据。这时候需要先把存量数据导一遍。

我常用的做法是，先建立一个临时表记录全量导出的位点，然后用SELECT分批把 MySQL 全量数据读出，写入 ES，完成之后再启动 Canal 消费增量日志。之所以要记录位点，是为了避免全量导出的过程中业务又发生了新的变更，导致这部分变更在 Canal 启动前没被消费。

更简单的方案是直接用 ES 官方提供的工具做全量迁移，但要注意：如果全量导入和 Canal 增量消费同时进行，可能会因为_id相同导致相互覆盖。稳妥的做法是先暂停业务写入或者接受短时间的数据不一致，全量导入完成后立即启动 Canal，在低峰期操作的话影响一般可控。

5. 常见问题与排查技巧：Canal 同步链路的避坑指南

5.1 MySQL 连接不上或认证失败

这是最常遇到的问题，尤其是 MySQL 8.0。Canal 日志里出现Access denied for user 'canal'@'xxx'时，先用下面的 SQL 检查账号权限：

SHOW GRANTS FOR 'canal'@'%';

如果没有REPLICATION SLAVE权限，重新执行授权。如果 MySQL 8.0 认证插件不兼容，需要把账号改成mysql_native_password。另外还有一个隐蔽问题： Canal 依赖连接 MySQL 时的字符集设置，如果你的库表用了 utf8mb4，canal.instance.connectionCharset最好也设置成UTF-8而不是GBK，否则中文会出现乱码。

5.2 启动后没有消费到数据

Canal 启动正常，但改了 MySQL 数据后消费端没有任何反应，这种情况多半是位点或订阅规则的问题。先在 MySQL 端手动执行一条 UPDATE，然后看example.log里有没有解析记录。

如果没有，检查canal.instance.filter.regex的写法。这个配置项里的\\是 Java 正则表达式的转义写法，test_db.user表示只订阅单表，如果你写成了test_db\.user（只有一个反斜杠），在 Java 属性文件里可能被非法转义，导致规则失效。

另一个可能的原因是 Canal 默认从启动时刻开始监听，如果 MySQL 里的变更发生在 Canal 启动之前，自然收不到。想验证这个问题，可以在 MySQL 端插入一条新数据，再看看消费端。

5.3 消费时重复或丢失

Canal 的消费模型是“至少一次”，也就是说极端情况下，消息可能被重复消费。程序崩溃后重启，如果ack没来得及发送，Canal 会从最后未确认的位点重新推送，这就会造成重复。

好在我们的消费端写入 ES 时用的是 MySQL 主键作为_id，重复消费同一个 INSERT 事件也只是覆盖写同一个文档，最终结果一致。对于 UPDATE 事件也一样。所以只要坚持用主键作为_id，重复消费基本无害。

但如果生产环境有严格的顺序要求，比如同一个 id 的 INSERT 和 DELETE 事件先后到达，而消费端是并发处理的，就可能出现先删后插的问题。解决办法有几种：让同一个 id 的数据进入同一个线程处理，或者用 ES 的版本号机制控制写入，或者消费端引入去重表。

5.4 ES 写入性能优化

当 MySQL 的变更量很大时，Canal 消费端逐条写入 ES 的性能会成为瓶颈。通常我会做两件事：

第一，消费端批量写入。Canal 一次getWithoutAck可以拿到很多条事件，把它们攒成一个列表，调用 ES 的BulkRequest一次性写入，吞吐量能提升一个量级。代码里把batchSize调大，比如 500 或 1000，处理完一批再ack。

第二，调整 Canal 的内存配置。canal.instance.memory.buffer.size控制了解析事件的内存缓冲大小，默认 16384 表示最多缓冲 16384 条事件。如果消费端处理不过来，缓冲区满了，Canal 会阻塞在获取事件阶段，间接限流。这时候要评估是消费端太慢还是缓冲区太小，对症下药。

5.5 MySQL 主从切换或 Canal 重启后位点失效

Canal 会把消费位点记录在本地，具体路径是conf/example/meta.dat。正常停机重启没问题，但如果在 Canal 停止期间 MySQL 的 binlog 被清理了，或者发生了主从切换，Canal 重启后可能找不到对应的位点文件，报错Could not find first log file name in binary log index。

遇到这种情况，最简单的处理方式是删除meta.dat，让 Canal 从当前最新的 binlog 位置重新开始监听。但这意味着中间漏掉的数据不会再补了，所以删除前要想清楚，或者先用全量方式把 ES 数据重新导一遍。

5.6 大事务或批量操作导致的事件风暴

MySQL 一条 UPDATE 语句如果影响了一万行数据，Canal 会生成一万条 RowData 事件。如果业务下游对这些事件一一处理，瞬间压力会很大。

我的建议是，在消费端做两层防护。第一层是缓冲限流，不要无限接收事件，用有界队列或批量窗口控制写入 ES 的速率；第二层是业务层面的合并，如果一个事务变更了同一个 id 的多条历史记录，可以只保留最终状态写入 ES，减少不必要的写操作。Canal 事件里带了事务 ID，可以根据事务 ID 做一定粒度的聚合。

5.7 常见问题速查表

5.8 几个生产环境的补充建议

部署形态上，如果 Canal 单机故障会导致整个同步链路中断，有条件的话建议部署 Canal 集群，配合 Zookeeper 选主，实现高可用。Zookeeper 会记录每个 Canal 实例的位点，主节点故障后从节点可以接着消费，不会丢数据。

网络层面，Canal 与 MySQL 之间、Canal 与 ES 之间的网络延迟直接影响同步延迟。如果跨机房部署，尽量保证三者在同一个内网，避免公网传输导致的抖动和延迟。

监控层面，除了看 Canal 日志外，还可以关注几个指标：Canal Server 的延迟时间（可以通过日志里的nextPosition和当前时间估算）、消费端getWithoutAck的返回条数、ES 的写入延迟和拒绝率。把这些指标接到 Prometheus 或者日志平台里，出现问题能第一时间发现。

6. 写在最后的实操心得

这套方案跑下来，我最大的一个感受是：Canal 真正的价值不只是把 MySQL 数据搬进 ES，而是给业务提供了一条稳定、可观测、可回溯的数据变更管道。搞明白了 binlog 和位点机制之后，你会发现很多原来需要写脚本轮询的任务都可以换成交替事件驱动的方式来做，不只是 ES，比如缓存更新、数据审计、异构数据同步，都是同一套思路。

如果是从零开始接触这个链路，我建议你先别急着上复杂架构。先在自己电脑上把 MySQL、Canal、ES 三者都装起来，按文章里的配置跑通一个最简单的例子，再逐步往里面加批量、加集群。有个小技巧是我后来才发现的：调试阶段可以在 MySQL 端开启 general_log，这样能清楚看到 Canal 伪装从库时的连接行为，很多“为什么没同步”的疑问都会迎刃而解。等你把这条链路跑顺了，再碰到数据同步的需求，就会有一种手里有粮、心里不慌的感觉。