MySQL同步ES的六種方案！
作者：蘇三
這篇文章跟大家一起聊聊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雙保險
責任編輯：武曉燕
來源：
蘇三學技術
