面試官：如何保證消息的可靠性傳輸？或者說，如何處理消息丟失的問題？

本文由 yanglbme 首發於 GitHub 技術社區 Doocs，目前 stars 已超 30k。
項目地址：github.com/doocs/advan…

面試題

如何保證消息的可靠性傳輸？或者說，如何處理消息丟失的問題？

面試官心理分析

這個是確定的，用 MQ 有個基本原則，就是數據不能多一條，也不能少一條，不能多，就是前面說的重複消費和冪等性問題。不能少，就是說這數據別搞丟了。那這個問題你必須得考慮一下。

若是說你這個是用 MQ 來傳遞很是核心的消息，好比說計費、扣費的一些消息，那必須確保這個 MQ 傳遞過程當中絕對不會把計費消息給弄丟。

面試題剖析

數據的丟失問題，可能出如今生產者、MQ、消費者中，我們從 RabbitMQ 和 Kafka 分別來分析一下吧。

RabbitMQ

生產者弄丟了數據

生產者將數據發送到 RabbitMQ 的時候，可能數據就在半路給搞丟了，由於網絡問題啥的，都有可能。

此時能夠選擇用 RabbitMQ 提供的事務功能，就是生產者發送數據以前開啓 RabbitMQ 事務channel.txSelect，而後發送消息，若是消息沒有成功被 RabbitMQ 接收到，那麼生產者會收到異常報錯，此時就能夠回滾事務channel.txRollback，而後重試發送消息；若是收到了消息，那麼能夠提交事務channel.txCommit。

// 開啓事務
channel.txSelect
try {
    // 這裏發送消息
} catch (Exception e) {
    channel.txRollback

    // 這裏再次重發這條消息
}

// 提交事務
channel.txCommit
複製代碼

可是問題是，RabbitMQ 事務機制（同步）一搞，基本上吞吐量會下來，由於太耗性能。

因此通常來講，若是你要確保說寫 RabbitMQ 的消息別丟，能夠開啓 confirm 模式，在生產者那裏設置開啓 confirm 模式以後，你每次寫的消息都會分配一個惟一的 id，而後若是寫入了 RabbitMQ 中，RabbitMQ 會給你回傳一個 ack 消息，告訴你說這個消息 ok 了。若是 RabbitMQ 沒能處理這個消息，會回調你的一個 nack 接口，告訴你這個消息接收失敗，你能夠重試。並且你能夠結合這個機制本身在內存裏維護每一個消息 id 的狀態，若是超過必定時間還沒接收到這個消息的回調，那麼你能夠重發。

事務機制和 confirm 機制最大的不一樣在於，事務機制是同步的，你提交一個事務以後會阻塞在那兒，可是 confirm 機制是異步的，你發送個消息以後就能夠發送下一個消息，而後那個消息 RabbitMQ 接收了以後會異步回調你的一個接口通知你這個消息接收到了。

因此通常在生產者這塊避免數據丟失，都是用 confirm 機制的。

RabbitMQ 弄丟了數據

就是 RabbitMQ 本身弄丟了數據，這個你必須開啓 RabbitMQ 的持久化，就是消息寫入以後會持久化到磁盤，哪怕是 RabbitMQ 本身掛了，恢復以後會自動讀取以前存儲的數據，通常數據不會丟。除非極其罕見的是，RabbitMQ 還沒持久化，本身就掛了，可能致使少許數據丟失，可是這個機率較小。

設置持久化有兩個步驟：

• 建立 queue 的時候將其設置爲持久化
  這樣就能夠保證 RabbitMQ 持久化 queue 的元數據，可是它是不會持久化 queue 裏的數據的。
• 第二個是發送消息的時候將消息的 deliveryMode 設置爲 2
  就是將消息設置爲持久化的，此時 RabbitMQ 就會將消息持久化到磁盤上去。

必需要同時設置這兩個持久化才行，RabbitMQ 哪怕是掛了，再次重啓，也會從磁盤上重啓恢復 queue，恢復這個 queue 裏的數據。

注意，哪怕是你給 RabbitMQ 開啓了持久化機制，也有一種可能，就是這個消息寫到了 RabbitMQ 中，可是還沒來得及持久化到磁盤上，結果不巧，此時 RabbitMQ 掛了，就會致使內存裏的一點點數據丟失。

因此，持久化能夠跟生產者那邊的 confirm 機制配合起來，只有消息被持久化到磁盤以後，纔會通知生產者 ack 了，因此哪怕是在持久化到磁盤以前，RabbitMQ 掛了，數據丟了，生產者收不到 ack，你也是能夠本身重發的。

消費端弄丟了數據

RabbitMQ 若是丟失了數據，主要是由於你消費的時候，剛消費到，還沒處理，結果進程掛了，好比重啓了，那麼就尷尬了，RabbitMQ 認爲你都消費了，這數據就丟了。

這個時候得用 RabbitMQ 提供的 ack 機制，簡單來講，就是你必須關閉 RabbitMQ 的自動 ack，能夠經過一個 api 來調用就行，而後每次你本身代碼裏確保處理完的時候，再在程序裏 ack 一把。這樣的話，若是你還沒處理完，不就沒有 ack 了？那 RabbitMQ 就認爲你還沒處理完，這個時候 RabbitMQ 會把這個消費分配給別的 consumer 去處理，消息是不會丟的。

Kafka

消費端弄丟了數據

惟一可能致使消費者弄丟數據的狀況，就是說，你消費到了這個消息，而後消費者那邊自動提交了 offset，讓 Kafka 覺得你已經消費好了這個消息，但其實你纔剛準備處理這個消息，你還沒處理，你本身就掛了，此時這條消息就丟咯。

這不是跟 RabbitMQ 差很少嗎，你們都知道 Kafka 會自動提交 offset，那麼只要關閉自動提交 offset，在處理完以後本身手動提交 offset，就能夠保證數據不會丟。可是此時確實仍是可能會有重複消費，好比你剛處理完，還沒提交 offset，結果本身掛了，此時確定會重複消費一次，本身保證冪等性就行了。

生產環境碰到的一個問題，就是說咱們的 Kafka 消費者消費到了數據以後是寫到一個內存的 queue 裏先緩衝一下，結果有的時候，你剛把消息寫入內存 queue，而後消費者會自動提交 offset。而後此時咱們重啓了系統，就會致使內存 queue 裏還沒來得及處理的數據就丟失了。

Kafka 弄丟了數據

這塊比較常見的一個場景，就是 Kafka 某個 broker 宕機，而後從新選舉 partition 的 leader。你們想一想，要是此時其餘的 follower 恰好還有些數據沒有同步，結果此時 leader 掛了，而後選舉某個 follower 成 leader 以後，不就少了一些數據？這就丟了一些數據啊。

生產環境也遇到過，咱們也是，以前 Kafka 的 leader 機器宕機了，將 follower 切換爲 leader 以後，就會發現說這個數據就丟了。

因此此時通常是要求起碼設置以下 4 個參數：

• 給 topic 設置 replication.factor 參數：這個值必須大於 1，要求每一個 partition 必須有至少 2 個副本。
• 在 Kafka 服務端設置 min.insync.replicas 參數：這個值必須大於 1，這個是要求一個 leader 至少感知到有至少一個 follower 還跟本身保持聯繫，沒掉隊，這樣才能確保 leader 掛了還有一個 follower 吧。
• 在 producer 端設置 acks=all：這個是要求每條數據，必須是寫入全部 replica 以後，才能認爲是寫成功了。
• 在 producer 端設置 retries=MAX（很大很大很大的一個值，無限次重試的意思）：這個是要求一旦寫入失敗，就無限重試，卡在這裏了。

咱們生產環境就是按照上述要求配置的，這樣配置以後，至少在 Kafka broker 端就能夠保證在 leader 所在 broker 發生故障，進行 leader 切換時，數據不會丟失。

生產者會不會弄丟數據？

若是按照上述的思路設置了 acks=all，必定不會丟，要求是，你的 leader 接收到消息，全部的 follower 都同步到了消息以後，才認爲本次寫成功了。若是沒知足這個條件，生產者會自動不斷的重試，重試無限次。

---

歡迎關注個人微信公衆號「Doocs開源社區」，原創技術文章第一時間推送。