MySQL同步Elasticsearch的6种方案小结

魁首哥

作者

引言

在分布式架构中，mysql与elasticsearch（es）的协同已成为解决高并发查询与复杂检索的标配组合。

然而，如何实现两者间的高效数据同步，是架构设计中绕不开的难题。

这篇文章跟大家一起聊聊mysql同步es的6种主流方案，结合代码示例与场景案例，帮助开发者避开常见陷阱，做出最优技术选型。

方案一：同步双写

场景：适用于对数据实时性要求极高，且业务逻辑简单的场景，如金融交易记录同步。

在业务代码中同时写入mysql与es。

代码如下：

@transactional  
public void createorder(order order) {  
    // 写入mysql  
    ordermapper.insert(order);  
    // 同步写入es  
    indexrequest request = new indexrequest("orders")  
        .id(order.getid())  
        .source(json.tojsonstring(order), xcontenttype.json);  
    client.index(request, requestoptions.default);  
}

痛点：

硬编码侵入：所有涉及写操作的地方均需添加es写入逻辑。
性能瓶颈：双写操作导致事务时间延长，tps下降30%以上。
数据一致性风险：若es写入失败，需引入补偿机制（如本地事务表+定时重试）。

方案二：异步双写

场景：电商订单状态更新后需同步至es供客服系统检索。

我们可以使用mq进行解耦。

架构图如下：

代码示例如下：

// 生产者端  
public void updateproduct(product product) {  
    productmapper.update(product);  
    kafkatemplate.send("product-update", product.getid());  
}  

// 消费者端  
@kafkalistener(topics = "product-update")  
public void synctoes(string productid) {  
    product product = productmapper.selectbyid(productid);  
    esclient.index(product);  
}

优势：

吞吐量提升：通过mq削峰填谷，可承载万级qps。
故障隔离：es宕机不影响主业务链路。

缺陷：

消息堆积：突发流量可能导致消费延迟（需监控lag值）。
顺序性问题：需通过分区键保证同一数据的顺序消费。

方案三：logstash定时拉取

场景：用户行为日志的t+1分析场景。

该方案低侵入但高延迟。

配置示例如下：

input {  
  jdbc {  
    jdbc_driver => "com.mysql.jdbc.driver"  
    jdbc_url => "jdbc:mysql://localhost:3306/log_db"  
    schedule => "*/5 * * * *"  # 每5分钟执行  
    statement => "select * from user_log where update_time > :sql_last_value"  
  }  
}  
output {  
  elasticsearch {  
    hosts => ["es-host:9200"]  
    index => "user_logs"  
  }  
}

适用性分析：

优点：零代码改造，适合历史数据迁移。
致命伤：
- 分钟级延迟（无法满足实时搜索）
- 全表扫描压力大（需优化增量字段索引）

方案四：canal监听binlog

场景：社交平台动态实时搜索（如微博热搜更新）。

技术栈：canal + rocketmq + es

该方案高实时，并且低侵入。

架构流程如下：

关键配置：

# canal.properties  
canal.instance.master.address=127.0.0.1:3306  
canal.mq.topic=canal.es.sync

避坑指南：

数据漂移：需处理ddl变更（通过schema registry管理映射）。
幂等消费：通过_id唯一键避免重复写入。

方案五：datax批量同步

场景：将历史订单数据从分库分表mysql迁移至es。

该方案是大数据迁移的首选。

配置文件如下：

{  
  "job": {  
    "content": [{  
      "reader": {  
        "name": "mysqlreader",  
        "parameter": { "splitpk": "id", "querysql": "select * from orders" }  
      },  
      "writer": {  
        "name": "elasticsearchwriter",  
        "parameter": { "endpoint": "http://es-host:9200", "index": "orders" }  
      }  
    }]  
  }  
}

性能调优：

调整channel数提升并发（建议与分片数对齐）
启用limit分批查询避免oom

方案六：flink流处理

场景：商品价格变更时，需关联用户画像计算实时推荐评分。

该方案适合于复杂的etl场景。

代码片段如下：

streamexecutionenvironment env = streamexecutionenvironment.getexecutionenvironment();  
env.addsource(new canalsource())  
   .map(record -> parsetopriceevent(record))  
   .keyby(event -> event.getproductid())  
   .connect(userprofilebroadcaststream)  
   .process(new pricerecommendationprocess())  
   .addsink(new elasticsearchsink());

优势：

状态管理：精准处理乱序事件（watermark机制）
维表关联：通过broadcast state实现实时画像关联

总结

对于文章上面给出的这6种技术方案，我们在实际工作中，该如何做选型呢？

下面用一张表格做对比：

方案	实时性	侵入性	复杂度	适用阶段
同步双写	秒级	高	低	小型单体项目
mq异步	秒级	中	中	中型分布式系统
logstash	分钟级	无	低	离线分析
canal	毫秒级	无	高	高并发生产环境
datax	小时级	无	中	历史数据迁移
flink	毫秒级	低	极高	实时数仓