引言

在分布式架構(gòu)中，MySQL與Elasticsearch（ES）的協(xié)同已成為解決高并發(fā)查詢與復(fù)雜檢索的標(biāo)配組合。

然而，如何實(shí)現(xiàn)兩者間的高效數(shù)據(jù)同步，是架構(gòu)設(shè)計(jì)中繞不開的難題。

這篇文章跟大家一起聊聊MySQL同步ES的6種主流方案，結(jié)合代碼示例與場景案例，幫助開發(fā)者避開常見陷阱，做出最優(yōu)技術(shù)選型。

方案一：同步雙寫

場景：適用于對(duì)數(shù)據(jù)實(shí)時(shí)性要求極高，且業(yè)務(wù)邏輯簡單的場景，如金融交易記錄同步。

在業(yè)務(wù)代碼中同時(shí)寫入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);  
}

痛點(diǎn)：

1. 硬編碼侵入：所有涉及寫操作的地方均需添加ES寫入邏輯。
2. 性能瓶頸：雙寫操作導(dǎo)致事務(wù)時(shí)間延長，TPS下降30%以上。
3. 數(shù)據(jù)一致性風(fēng)險(xiǎn)：若ES寫入失敗，需引入補(bǔ)償機(jī)制（如本地事務(wù)表+定時(shí)重試）。

方案二：異步雙寫

場景：電商訂單狀態(tài)更新后需同步至ES供客服系統(tǒng)檢索。

我們可以使用MQ進(jìn)行解耦。

架構(gòu)圖如下：

代碼示例如下：

// 生產(chǎn)者端  
public void updateProduct(Product product) {  
    productMapper.update(product);  
    kafkaTemplate.send("product-update", product.getId());  
}  

// 消費(fèi)者端  
@KafkaListener(topics = "product-update")  
public void syncToEs(String productId) {  
    Product product = productMapper.selectById(productId);  
    esClient.index(product);  
}

優(yōu)勢(shì)：

• 吞吐量提升：通過MQ削峰填谷，可承載萬級(jí)QPS。
• 故障隔離：ES宕機(jī)不影響主業(yè)務(wù)鏈路。

缺陷：

• 消息堆積：突發(fā)流量可能導(dǎo)致消費(fèi)延遲（需監(jiān)控Lag值）。
• 順序性問題：需通過分區(qū)鍵保證同一數(shù)據(jù)的順序消費(fèi)。

方案三：Logstash定時(shí)拉取

場景：用戶行為日志的T+1分析場景。

該方案低侵入但高延遲。

配置示例如下：

input {
jdbc{
    jdbc_driver=>"com.mysql.jdbc.Driver"
    jdbc_url=>"jdbc:mysql://localhost:3306/log_db"
    schedule=>"*/5 * * * *"# 每5分鐘執(zhí)行  
    statement=>"SELECT * FROM user_log WHERE update_time > :sql_last_value"
}
}
output{
elasticsearch{
    hosts=>["es-host:9200"]
    index=>"user_logs"
}
}

適用性分析：

• 優(yōu)點(diǎn)：零代碼改造，適合歷史數(shù)據(jù)遷移。
• 致命傷：

• 分鐘級(jí)延遲（無法滿足實(shí)時(shí)搜索）
• 全表掃描壓力大（需優(yōu)化增量字段索引）

方案四：Canal監(jiān)聽Binlog

場景：社交平臺(tái)動(dòng)態(tài)實(shí)時(shí)搜索（如微博熱搜更新）。技術(shù)棧：Canal + RocketMQ + ES

該方案高實(shí)時(shí)，并且低侵入。

架構(gòu)流程如下：

關(guān)鍵配置：

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

避坑指南：

1. 數(shù)據(jù)漂移：需處理DDL變更（通過Schema Registry管理映射）。
2. 冪等消費(fèi)：通過_id唯一鍵避免重復(fù)寫入。

方案五：DataX批量同步

場景：將歷史訂單數(shù)據(jù)從分庫分表MySQL遷移至ES。

該方案是大數(shù)據(jù)遷移的首選。

配置文件如下：

{  
"job":{
    "content":[{
      "reader":{
        "name":"mysqlreader",
        "parameter":{"splitPk":"id","querySql":"SELECT * FROM orders"}
      },
      "writer":{
        "name":"elasticsearchwriter",
        "parameter":{"endpoint":"http://es-host:9200","index":"orders"}
      }
    }]
}
}

性能調(diào)優(yōu)：

• 調(diào)整channel數(shù)提升并發(fā)（建議與分片數(shù)對(duì)齊）
• 啟用limit分批查詢避免OOM

方案六：Flink流處理

場景：商品價(jià)格變更時(shí)，需關(guān)聯(lián)用戶畫像計(jì)算實(shí)時(shí)推薦評(píng)分。

該方案適合于復(fù)雜的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());

優(yōu)勢(shì)：

• 狀態(tài)管理：精準(zhǔn)處理亂序事件（Watermark機(jī)制）
• 維表關(guān)聯(lián)：通過Broadcast State實(shí)現(xiàn)實(shí)時(shí)畫像關(guān)聯(lián)

總結(jié)：

對(duì)于文章上面給出的這6種技術(shù)方案，我們?cè)趯?shí)際工作中，該如何做選型呢？

下面用一張表格做對(duì)比：

建議：

1. 若團(tuán)隊(duì)無運(yùn)維中間件能力 → 選擇Logstash或同步雙寫
2. 需秒級(jí)延遲且允許改造 → MQ異步 + 本地事務(wù)表
3. 追求極致實(shí)時(shí)且資源充足 → Canal + Flink雙保險(xiǎn)