Elasticsearch 与 MySQL 数据同步踩坑

同步方案不止一种

MySQL 是事实来源，Elasticsearch 是查询副本，两者之间必然存在延迟。同步方案的选择，本质上是在“实时性、一致性、复杂度”之间做取舍。

订单、商品这类要求秒级搜索的场景，主流做法是 Canal 订阅 MySQL binlog，投递到 MQ，再由消费者批量写入 ES。业务代码只写 MySQL，不感知 ES，链路失败时还能靠 MQ 重试和全量重建兜底。

代码：从变更消息到批量写入

消费端不要一条消息写一次 ES，批量提交能显著降低段合并和网络开销。Elasticsearch 8.x 的 Java Client 写法如下：

@Component
public class ProductSyncConsumer {

    private static final Logger log = LoggerFactory.getLogger(ProductSyncConsumer.class);

    private final ElasticsearchClient esClient;

    public ProductSyncConsumer(ElasticsearchClient esClient) {
        this.esClient = esClient;
    }

    @RabbitListener(queues = "es.product.sync.queue")
    public void onMessage(ProductChangeMessage message) throws IOException {
        ProductDoc doc = message.toDoc();
        BulkRequest.Builder builder = new BulkRequest.Builder();
        builder.operations(op -> op
                .update(update -> update
                        .index("products")
                        .id(doc.id())
                        .document(doc)
                        .docAsUpsert(true)));
        BulkResponse response = esClient.bulk(builder.build());
        if (response.errors()) {
            response.items().stream()
                    .filter(item -> item.error() != null)
                    .forEach(item -> log.error("ES 写入失败, id={}, reason={}",
                            item.id(), item.error().reason()));
            throw new IllegalStateException("部分文档写入失败，触发 MQ 重试");
        }
    }
}

docAsUpsert(true) 让“更新不存在就插入”成为原子操作，避免先查再写。整批失败直接抛异常，让 MQ 重新投递；部分失败要记录失败 ID，不能因为大部分成功就认为任务完成。

常见坑

1. 消息乱序，旧值覆盖新值

binlog 变更按主键分发到同一个 MQ 分区，消费端再按主键串行处理，才能保证同一文档的顺序。如果 MQ 是轮询分发、消费端又是多线程，后发的旧消息可能覆盖新值。更稳的做法是给文档加 version 字段，写入时用外部版本号做乐观并发控制。

2. 删除被当成“更新为空”

MySQL 物理删除后，binlog 里是 DELETE 事件，消费端要执行 ES 的 delete，而不是把文档字段更新成 null。逻辑删除则要同步 deleted 标记，并在查询时过滤。两种策略要统一，不能一半物理删、一半逻辑删。

3. 字段类型对不上

MySQL 的 DECIMAL(10,2) 同步到 ES 用 double 会丢精度，应该用 scaled_float 或 keyword；DATETIME 没有时区，同步时要明确按哪个时区转成 ISO 8601；MySQL 允许 null，ES 的 date 字段收到空字符串会解析失败。映射不是复制字段名，而是逐字段确认类型。

4. 部分更新把字段冲掉

用全量文档覆盖时，如果 MySQL 查询漏了字段，ES 里的对应字段会被置空。要么每次取完整行，要么用 partial update 并明确字段列表。doc_as_upsert 只解决“不存在时插入”，不解决“字段缺失”。

5. 写完立刻搜不到

ES 默认每秒 refresh，binlog 同步完成后立刻查可能还是旧结果。这是设计上的近实时，不是同步失败。对“写完必须立刻读到”的场景，可以让查询走 MySQL，或者写入时指定 refresh=wait_for，但要评估对写入吞吐的影响。

6. 全量重建时把线上索引写脏

直接往线上索引灌全量数据，期间新旧数据混在一起，用户会看到重复或错误结果。正确做法是新建索引、灌数据、校验、原子切换别名；切换前保留旧索引，出问题能快速回滚。

一致性校验

同步链路再可靠，也要有对账。轻量做法是定时抽样：按主键取一批 MySQL 记录，逐条查 ES 的 _source，比较关键字段和更新时间。全量做法是按天跑一次批处理，比较两边 count、最大值、校验和，发现差异记录到修复表。

修复任务要幂等：以 MySQL 当前值为准，重新写入 ES；删除类差异以 MySQL 是否存在为准。修复过程中同样要防止旧消息回流，否则刚修好的数据又被覆盖。

排查步骤

先确认文档在不在、内容对不对：

curl -u elastic:password "http://localhost:9200/products/_doc/1001?pretty"
curl -u elastic:password "http://localhost:9200/products/_count?pretty"
curl -u elastic:password "http://localhost:9200/_cat/indices/products?v"

如果文档缺失，往上游查：Canal 是否还在推进位点、MQ 是否有积压、消费端有没有批量失败。ES 写入被拒时，看线程池和磁盘水位：

curl -u elastic:password "http://localhost:9200/_cat/thread_pool/write?v"
curl -u elastic:password "http://localhost:9200/_cluster/health?pretty"

write 线程池的 rejected 持续增长，通常是批量太大或磁盘达到 flood stage；先降批量、加副本，再谈扩容。

小结

MySQL 到 ES 的同步，核心不是“怎么把数据写过去”，而是“怎么处理失败、重复、乱序和删除”。链路要有重试，消费要幂等，写入要能对账，索引要能重建。把这四件事设计好，同步延迟反而成了最容易接受的问题。

参考

• RabbitMQ 官方文档
• 博客园：RabbitMQ 消息丢失怎么解决
• RocketMQ 官方文档
• CSDN：RocketMQ / RabbitMQ 深度解析
• Elasticsearch 官方指南
• 博客园：ES 倒排索引实战
• 博客园：Kafka 消费者组订阅问题