Canal 监听 binlog 到 ES：把同步延迟从 5 分钟压到秒级

一、业务背景：为什么不用定时任务同步搜索数据

电商商品表在 MySQL 中承担事务写入，但商品搜索需要按标题、类目、品牌、价格区间、标签做多维检索。早期方案用每 5 分钟一次的定时任务扫描 update_time，再批量重建 Elasticsearch 索引。随着商品量涨到千万级，问题开始集中爆发：全量扫描拖慢主库、增量漏数频繁、搜索可见延迟长期在分钟级，运营改价后客服不断收到“为什么搜不到新价格”的投诉。

更合理的做法是把 MySQL 的 binlog 作为事实源：业务库只负责可靠写入，Canal 伪装成 MySQL 从库订阅变更，经过消息队列削峰与解耦，再由消费者把行变更转换成 ES 文档。这样既能做到秒级索引可见，也能把同步压力从主库剥离。

二、方案对比：轮询、双写与 binlog 订阅

常见同步方案主要有三类：

• 业务双写：代码里同时写 MySQL 和 ES，延迟最低，但分布式事务复杂，ES 故障会影响下单主链路，且历史代码改造面大。
• 定时任务轮询：实现最简单，但依赖 update_time，无法感知删除事件，扫描大表成本高，延迟和漏数问题明显。
• Canal 订阅 binlog：对业务代码几乎零侵入，能拿到 insert、update、delete 的完整行镜像，适合构建搜索、缓存、宽表等衍生视图。

binlog 方案的代价是链路更长，需要保证消息有序、消费幂等、失败可重试。因此生产环境通常不会让 Canal 直接写 ES，而是加入消息队列承接流量。

三、MySQL 与 Canal 核心配置

先开启 MySQL 的 ROW 模式 binlog。ROW 会记录变更前后的完整行数据，这是把数据库行准确转换成 ES 文档的前提。

# my.cnf
[mysqld]
log-bin=mysql-bin
binlog_format=ROW
binlog_row_image=FULL
server-id=1

创建 Canal 所需账号，并授予复制权限：

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

Canal 的 instance.properties 至少要配置如下内容：

canal.instance.master.address=127.0.0.1:3306
canal.instance.dbUsername=canal
canal.instance.dbPassword=canal123
canal.instance.connectionCharset=UTF-8
# 订阅商品库的商品表与库存表
canal.instance.filter.regex=product_db\\.(product|inventory)
# 投递到消息队列时建议按主键分区，避免同一商品乱序
canal.mq.partition=0
canal.mq.partitionsNum=6
canal.mq.partitionHash=product_db.product:id

消息体中最关键的是表名、事件类型、主键和变更后列值。下面示例不依赖具体 MQ SDK，先用 Java 类表达统一的变更事件，Kafka、RocketMQ 或 RabbitMQ 都可以复用这层模型。

四、Java 消费者：把 binlog 事件写入 ES

引入 Elasticsearch Java Client：

<dependency>
    <groupId>co.elastic.clients</groupId>
    <artifactId>elasticsearch-java</artifactId>
    <version>8.13.4</version>
</dependency>

定义变更事件与商品文档：

// binlog 行变更事件，字段与 MQ 中的 JSON 对应
public record BinlogEvent(
        String database,
        String table,
        String type,          // INSERT / UPDATE / DELETE
        Long id,
        Map<String, Object> after,
        Long executeTime
) {}

// Elasticsearch 中的商品文档
public record ProductDoc(
        Long id,
        String title,
        Long categoryId,
        Long brandId,
        Integer price,
        Integer stock,
        Long updatedAt
) {}

同步服务负责事件转换和 ES 写入。关键点是使用商品主键作为文档 ID，重复消费时执行覆盖写，天然具备幂等性。

import co.elastic.clients.elasticsearch.ElasticsearchClient;
import co.elastic.clients.elasticsearch.core.IndexRequest;
import com.fasterxml.jackson.databind.ObjectMapper;
import java.util.Map;

public class ProductSyncService {

    private final ElasticsearchClient esClient;
    private final ObjectMapper objectMapper;

    public ProductSyncService(ElasticsearchClient esClient,
                              ObjectMapper objectMapper) {
        this.esClient = esClient;
        this.objectMapper = objectMapper;
    }

    public void sync(BinlogEvent event) throws Exception {
        // 只处理商品表；库存表可在聚合器中合并后再更新
        if (!"product".equals(event.table())) {
            return;
        }

        switch (event.type()) {
            case "INSERT", "UPDATE" -> indexProduct(event.after());
            case "DELETE" -> deleteProduct(event.id());
            default -> { /* DDL 等事件进入告警或忽略列表 */ }
        }
    }

    private void indexProduct(Map<String, Object> after) throws Exception {
        ProductDoc doc = new ProductDoc(
                toLong(after.get("id")),
                (String) after.get("title"),
                toLong(after.get("category_id")),
                toLong(after.get("brand_id")),
                toInt(after.get("price")),
                toInt(after.get("stock")),
                toLong(after.get("update_time"))
        );

        // 主键相同即覆盖，消费重试不会产生重复文档
        IndexRequest<ProductDoc> request = IndexRequest.of(b -> b
                .index("product_index")
                .id(String.valueOf(doc.id()))
                .document(doc)
        );
        esClient.index(request);
    }

    private void deleteProduct(Long id) throws Exception {
        esClient.delete(d -> d.index("product_index").id(String.valueOf(id)));
    }

    private Long toLong(Object value) {
        return value == null ? null : Long.valueOf(value.toString());
    }

    private Integer toInt(Object value) {
        return value == null ? null : Integer.valueOf(value.toString());
    }
}

实际项目中，商品信息与库存可能分散在不同表。消费者不要看到任意一行变更就立即覆盖整个 ES 文档，否则可能用旧数据把新字段回滚。更稳妥的方式是按商品 ID 做短时间窗口聚合，收到变更后回源查询最新聚合数据，再一次性写入。

public ProductDoc loadLatestProduct(Long productId) {
    // 事务中读取商品、库存、品牌等最新状态
    // 注意从从库读取可能有主从延迟，关键场景可强制走主库或等待安全位点
    return productMapper.selectAggregatedProduct(productId);
}

索引映射建议提前创建，避免动态映射把价格、状态等数字识别错误：

PUT /product_index
{
  "mappings": {
    "properties": {
      "id": { "type": "long" },
      "title": { "type": "text", "analyzer": "ik_max_word" },
      "categoryId": { "type": "long" },
      "brandId": { "type": "long" },
      "price": { "type": "integer" },
      "stock": { "type": "integer" },
      "updatedAt": { "type": "long" }
    }
  }
}

五、顺序、重试与全量初始化

顺序问题要按“同一主键有序”处理，而不是追求全局有序。MQ 分区键使用商品 ID，Canal 到 MQ、消费者到 ES 都保持同一 ID 落在同一分区，即可避免旧事件覆盖新事件。跨表聚合场景还可以在文档中保存 updatedAt 或 binlog 位点，写入前判断版本。

失败重试不能无限阻塞消费线程。建议采用三次指数退避，仍失败则投递死信队列并告警，由补偿任务按主键回源修复。监控至少覆盖消费延迟、死信数量、Canal 位点延迟、ES 写入拒绝数和主键不一致抽样结果。

首次上线时需要全量初始化：先记录当前 binlog 位点，再通过分批游标把历史数据写入 ES，随后从记录位点开始消费增量。若先启动增量再跑全量，重复数据虽然可被文档 ID 覆盖，但删除事件可能在全量任务之后到达，导致已删除商品重新出现。

六、避坑总结

1. 只收到变更列，构建文档时字段丢失：常见原因是 binlog_row_image 被设置为 MINIMAL，UPDATE 事件只包含主键和被修改列，消费者却直接用 after 覆盖整个 ES 文档，最终标题、类目变成空值。修正方式是设置为 FULL，或在消费端按主键回源聚合后再写。
2. 多表更新顺序错乱导致旧数据回滚：库存表和商品表分别发消息，消费者并发处理时，旧库存可能晚于新库存写入。修正方式是按商品主键分区、单分区顺序消费，并在 ES 文档中保留版本号或更新时间，低版本事件直接丢弃。
3. DELETE 事件被误写成空文档：把删除也走统一 index 方法，结果 ES 中留下只有 ID 的文档，搜索过滤异常。修正方式是区分事件类型，DELETE 必须调用 delete，业务软删除则应在查询过滤中增加 status 条件。
4. 字段命名和类型不一致引发查询失效：binlog 中是 category_id，Java 记录中是 categoryId，动态映射还可能把长整型识别成字符串。修正方式是固定显式 mapping，并在转换层集中维护字段映射，不要依赖 ES 自动推断。
5. 消费堆积拖垮 ES：大促批量改价时逐条写入，ES 线程池打满并出现拒绝。修正方式是在保证同主键顺序的前提下使用 Bulk 批量提交，给同步消费者独立限流，并通过死信队列承接异常数据。

Canal + binlog + MQ + Elasticsearch 并不是“装上就能忘”的管道，但只要把 ROW 配置、主键有序、幂等覆盖、死信补偿和全量增量衔接处理好，就能把搜索数据同步从分钟级轮询升级为稳定的秒级管道，同时保持业务写入链路简单可靠。

免责声明

本工具与文章内容仅用于学习测试场景，优化建议仅供参考，请勿直接用于生产环境，由此产生的风险自行承担。