思維導(dǎo)圖

0. 前言

MySQL 與 Elasticsearch 一致性問題是老生常談了。網(wǎng)上有太多關(guān)于這方面的文章了，但是千篇一律，看了跟沒看沒有太大區(qū)別。
在生產(chǎn)中，我們往往會通過 DTS 工具將 binlog 導(dǎo)入到 Kafka，再通過 Kafka 消費(fèi) binlog，組裝數(shù)據(jù)寫入 ES。
在這個過程中，可能會存在 binlog 到 Kafka 數(shù)據(jù)丟失，或者應(yīng)用程序消費(fèi) Kafka 數(shù)據(jù)成功，但是數(shù)據(jù)未正確寫入至 ES。
本文將探討如何解決 MySQL 與 ES 的數(shù)據(jù)一致性問題。

1. 消費(fèi)側(cè) Client 到 ES 數(shù)據(jù)丟失

這里出現(xiàn)丟失的問題，可能有以下幾種情況

1. ES 寫入線程滿了，請求被拒絕。
2. ES 寫入沖突。

1.1 ES 寫入請求被拒絕

一般是通過遞階式重試來解決問題，例如第一次等待 1s 后寫入，第二次還出現(xiàn)，則等待 3s 后再嘗試寫入。如果最后寫入還是失敗，應(yīng)該記錄日志，并告警，而后通過人工介入的方式補(bǔ)償數(shù)據(jù)。

不過也需要考慮另外一個問題，為什么并發(fā)這么高？這么高的寫入并發(fā)，對 ES 壓力是否太大了？

一般而言，我們認(rèn)為 ES 是不適合并發(fā)太高的寫入。因此在消費(fèi)側(cè)除了要控制 MQ 并發(fā)消費(fèi)的線程數(shù)，也要多用用 同步 Bulk API 做批量更新。

1.2 ES 寫入沖突

消費(fèi)側(cè)的邏輯一般如下：

1. 會將有關(guān)聯(lián)記錄打到同一個隊(duì)列，防止并發(fā)問題。例如同一個商品的 Binlog 都打到同一個隊(duì)列
2. 在主表的 Insert Binlog 中，查詢關(guān)聯(lián)表信息，拼裝完整記錄寫入 ES
3. 其它關(guān)聯(lián)表的更新、寫入以及主表的 Update Binlog 都用 Update API 做部分更新操作
4. Delete Binlog 用 Delete API 做刪除操作

如果是通過我上面說的方式進(jìn)行寫入，會出現(xiàn)沖突問題的僅有 Update API。

ES Client Update Api 提供了 retry_on_conflict 參數(shù)。該參數(shù)的意思是，如果發(fā)生版本沖突則重試，該參數(shù)默認(rèn)為 0，即默認(rèn)不重試。生產(chǎn)環(huán)境中，我們可以通過配置中心動態(tài)配置該參數(shù)值。如果重試之后還是發(fā)生錯誤，建議捕獲版本沖突異常，并告警，然后人工手動更新。

2. DTS 工具到 Kafka 數(shù)據(jù)丟失

這里的丟失包含 2 個部分

1. DTS 到 Kafka 丟失
2. 數(shù)據(jù)在 Kafka Broker 端丟失

一般是采用定時增量校驗(yàn)。校驗(yàn) MySQL 和 Elasticsearch 數(shù)據(jù)是否一致性。
整體的流程圖如下所示：