MySQL同步ES的六種方案！

作者：蘇三 2025-05-06 00:00:05

數據庫 MySQL

這篇文章跟大家一起聊聊MySQL同步ES的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	毫秒級	低	極高	實時數倉

建議：

若團隊無運維中間件能力 → 選擇Logstash或同步雙寫
需秒級延遲且允許改造 → MQ異步 + 本地事務表
追求極致實時且資源充足 → Canal + Flink雙保險

責任編輯：武曉燕來源：蘇三學技術

MySQL ES 協同