引言

在分布式架構(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ù)選型。

方案一：同步雙寫

場景：適用于對數(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)：

• 硬編碼侵入：所有涉及寫操作的地方均需添加ES寫入邏輯。
• 性能瓶頸：雙寫操作導(dǎo)致事務(wù)時(shí)間延長，TPS下降30%以上。
• 數(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)勢：

• 吞吐量提升：通過MQ削峰填谷，可承載萬級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ù)遷移。
• 致命傷：
  • 分鐘級延遲（無法滿足實(shí)時(shí)搜索）
  • 全表掃描壓力大（需優(yōu)化增量字段索引）

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

場景：社交平臺動(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

避坑指南：

• 數(shù)據(jù)漂移：需處理DDL變更（通過Schema Registry管理映射）。
• 冪等消費(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ù)對齊）
• 啟用limit分批查詢避免OOM

方案六：Flink流處理

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

該方案適合于復(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)勢：

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

總結(jié)

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

下面用一張表格做對比：

蘇三的建議：

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

到此這篇關(guān)于MySQL同步Elasticsearch的6種方案小結(jié)的文章就介紹到這了,更多相關(guān)MySQL同步Elasticsearch內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

最新評論