## 寫在前面 又到了年底跳槽高峰季，很多小夥伴出去面試時，不少面試官都會問到訊息佇列的問題，不少小夥伴回答的不是很完美，有些小夥伴是心裡知道答案，嘴上卻沒有很好的表達出來，究其根本原因，還是對相關的知識點理解的不夠透徹。今天，我們就一起來探討下這個話題。注：文章有點長，你說你能一鼓作氣看完，我有點不信！！ 文章已收錄到： [https://github.com/sunshinelyz/technology-binghe](https://github.com/sunshinelyz/technology-binghe) [https://gitee.com/binghe001/technology-binghe](https://gitee.com/binghe001/technology-binghe) ## 什麼是訊息佇列？ 訊息佇列（Message Queue）是在訊息的傳輸過程中儲存訊息的容器，是應用間的通訊方式。訊息傳送後可以立即返回，由訊息系統保證訊息的可靠傳輸，訊息釋出者只管把訊息寫到佇列裡面而不用考慮誰需要訊息，而訊息的使用者也不需要知道誰釋出的訊息，只管到訊息佇列裡面取，這樣生產和消費便可以做到分離。 ## 為什麼要使用訊息佇列？ **優點：** - 非同步處理：例如簡訊通知、終端狀態推送、App推送、使用者註冊等 - 資料同步：業務資料推送同步 - 重試補償：記賬失敗重試 - 系統解耦：通訊上下行、終端異常監控、分散式事件中心 - 流量消峰：秒殺場景下的下單處理 - 釋出訂閱：HSF的服務狀態變化通知、分散式事件中心 - 高併發緩衝：日誌服務、監控上報 使用訊息佇列比較核心的作用就是：**解耦**、**非同步**、**削峰**。 **缺點：** - 系統可用性降低 系統引入的外部依賴越多，越容易掛掉？如何保證訊息佇列的高可用？ - 系統複雜度提高 怎麼保證訊息沒有重複消費？怎麼處理訊息丟失的情況？怎麼保證訊息傳遞的順序性？ - 一致性問題 A 系統處理完了直接返回成功了，人都以為你這個請求就成功了；但是問題是，要是 BCD 三個系統那裡，BD 兩個系統寫庫成功了，結果 C 系統寫庫失敗了，咋整？你這資料就不一致了。 以下主要討論的RabbitMQ和Kafka兩種訊息佇列。 ## 如何保證訊息佇列的高可用？ ### RabbitMQ的高可用 RabbitMQ的高可用是**基於主從**（非分散式）做高可用性。RabbitMQ 有三種模式：單機模式（Demo級別）、普通叢集模式（無高可用性）、映象叢集模式（高可用性）。 - **普通叢集模式** 普通叢集模式，意思就是在多臺機器上啟動多個 RabbitMQ 例項，每個機器啟動一個。你**建立的 queue，只會放在一個 RabbitMQ 例項上**，但是每個例項都同步 queue 的元資料（元資料可以認為是 queue 的一些配置資訊，通過元資料，可以找到 queue 所在例項）。你消費的時候，實際上如果連線到了另外一個例項，那麼那個例項會從 queue 所在例項上拉取資料過來。 這種方式確實很麻煩，也不怎麼好，**沒做到所謂的分散式**，就是個普通叢集。因為這導致你要麼消費者每次隨機連線一個例項然後拉取資料，要麼固定連線那個 queue 所在例項消費資料，前者有**資料拉取的開銷**，後者導致**單例項效能瓶頸**。 而且如果那個放 queue 的例項宕機了，會導致接下來其他例項就無法從那個例項拉取，如果你**開啟了訊息持久化**，讓 RabbitMQ 落地儲存訊息的話，**訊息不一定會丟**，得等這個例項恢復了，然後才可以繼續從這個 queue 拉取資料。 所以這個事兒就比較尷尬了，這就**沒有什麼所謂的高可用性**，**這方案主要是提高吞吐量的**，就是說讓叢集中多個節點來服務某個 queue 的讀寫操作。  * **映象叢集模式** 這種模式，才是所謂的 RabbitMQ 的高可用模式。跟普通叢集模式不一樣的是，在映象叢集模式下，你建立的 queue，無論元資料還是 queue 裡的訊息都會**存在於多個例項上**，就是說，每個 RabbitMQ 節點都有這個 queue 的一個**完整映象**，包含 queue 的全部資料的意思。然後每次你寫訊息到 queue 的時候，都會自動把**訊息同步**到多個例項的 queue 上。 那麼**如何開啟這個映象叢集模式**呢？其實很簡單，RabbitMQ 有很好的管理控制檯，就是在後臺新增一個策略，這個策略是**映象叢集模式的策略**，指定的時候是可以要求資料同步到所有節點的，也可以要求同步到指定數量的節點，再次建立 queue 的時候，應用這個策略，就會自動將資料同步到其他的節點上去了。 這樣的話，好處在於，你任何一個機器宕機了，沒事兒，其它機器（節點）還包含了這個 queue 的完整資料，別的 consumer 都可以到其它節點上去消費資料。壞處在於，第一，這個效能開銷也太大了吧，訊息需要同步到所有機器上，導致網路頻寬壓力和消耗很重！第二，這麼玩兒，不是分散式的，就**沒有擴充套件性可言**了，如果某個 queue 負載很重，你加機器，新增的機器也包含了這個 queue 的所有資料，並**沒有辦法線性擴充套件**你的 queue。你想，如果這個 queue 的資料量很大，大到這個機器上的容量無法容納了，此時該怎麼辦呢？  ### Kafka的高可用 Kafka 一個最基本的架構認識：由多個 broker 組成，每個 broker 是一個節點；你建立一個 topic，這個 topic 可以劃分為多個 partition，每個 partition 可以存在於不同的 broker 上，每個 partition 就放一部分資料。 這就是**天然的分散式訊息佇列**，就是說一個 topic 的資料，是**分散放在多個機器上的，每個機器就放一部分資料**。 實際上 RabbmitMQ 之類的，並不是分散式訊息佇列，它就是傳統的訊息佇列，只不過提供了一些叢集、HA(High Availability, 高可用性) 的機制而已，因為無論怎麼玩兒，RabbitMQ 一個 queue 的資料都是放在一個節點裡的，映象叢集下，也是每個節點都放這個 queue 的完整資料。 Kafka 0.8 以前，是沒有 HA 機制的，就是任何一個 broker 宕機了，那個 broker 上的 partition 就廢了，沒法寫也沒法讀，沒有什麼高可用性可言。 比如說，我們假設建立了一個 topic，指定其 partition 數量是 3 個，分別在三臺機器上。但是，如果第二臺機器宕機了，會導致這個 topic 的 1/3 的資料就丟了，因此這個是做不到高可用的。  Kafka 0.8 以後，提供了 HA 機制，就是 replica（複製品） 副本機制。每個 partition 的資料都會同步到其它機器上，形成自己的多個 replica 副本。所有 replica 會選舉一個 leader 出來，那麼生產和消費都跟這個 leader 打交道，然後其他 replica 就是 follower。寫的時候，leader 會負責把資料同步到所有 follower 上去，讀的時候就直接讀 leader 上的資料即可。只能讀寫 leader？很簡單，**要是你可以隨意讀寫每個 follower，那麼就要 care 資料一致性的問題**，系統複雜度太高，很容易出問題。Kafka 會均勻地將一個 partition 的所有 replica 分佈在不同的機器上，這樣才可以提高容錯性。  這麼搞，就有所謂的**高可用性**了，因為如果某個 broker 宕機了，沒事兒，那個 broker上面的 partition 在其他機器上都有副本的。如果這個宕機的 broker 上面有某個 partition 的 leader，那麼此時會從 follower 中**重新選舉**一個新的 leader 出來，大家繼續讀寫那個新的 leader 即可。這就有所謂的高可用性了。 **寫資料**的時候，生產者就寫 leader，然後 leader 將資料落地寫本地磁碟，接著其他 follower 自己主動從 leader 來 pull 資料。一旦所有 follower 同步好資料了，就會發送 ack 給 leader，leader 收到所有 follower 的 ack 之後，就會返回寫成功的訊息給生產者。（當然，這只是其中一種模式，還可以適當調整這個行為） **消費**的時候，只會從 leader 去讀，但是隻有當一個訊息已經被所有 follower 都同步成功返回 ack 的時候，這個訊息才會被消費者讀到。 ## 如何保證訊息不重複消費（冪等性）？ 首先，所有的訊息佇列都會有這樣重複消費的問題，因為這是不MQ來保證，而是我們自己開發保證的，我們使用Kakfa來討論是如何實現的。 Kakfa有個offset的概念，就是每個訊息寫進去都會有一個offset值，代表消費的序號，然後consumer消費了資料之後，預設每隔一段時間會把自己消費過的訊息的offset值提交，表示我已經消費過了，下次要是我重啟啥的，就讓我從當前提交的offset處來繼續消費。 但是凡事總有意外，比如我們之前生產經常遇到的，就是你有時候重啟系統，看你怎麼重啟了，如果碰到點著急的，直接 kill 程序了，再重啟。這會導致 consumer 有些訊息處理了，但是沒來得及提交 offset，尷尬了。重啟之後，少數訊息會再次消費一次。 其實重複消費不可怕，可怕的是你沒考慮到重複消費之後，**怎麼保證冪等性**。 舉個例子吧。假設你有個系統，消費一條訊息就往資料庫裡插入一條資料，要是你一個訊息重複兩次，你不就插入了兩條，這資料不就錯了？但是你要是消費到第二次的時候，自己判斷一下是否已經消費過了，若是就直接扔了，這樣不就保留了一條資料，從而保證了資料的正確性。一條資料重複出現兩次，資料庫裡就只有一條資料，這就保證了系統的冪等性。冪等性，通俗點說，就一個數據，或者一個請求，給你重複來多次，你得確保對應的資料是不會改變的，**不能出錯**。 所以第二個問題來了，怎麼保證訊息佇列消費的冪等性？ 其實還是得結合業務來思考，我這裡給幾個思路： - 比如你拿個資料要寫庫，你先根據主鍵查一下，如果這資料都有了，你就別插入了，update 一下好吧。 - 比如你是寫 Redis，那沒問題了，反正每次都是 set，天然冪等性。 - 比如你不是上面兩個場景，那做的稍微複雜一點，你需要讓生產者傳送每條資料的時候，裡面加一個全域性唯一的 id，類似訂單 id 之類的東西，然後你這裡消費到了之後，先根據這個 id 去比如 Redis 裡查一下，之前消費過嗎？如果沒有消費過，你就處理，然後這個 id 寫 Redis。如果消費過了，那你就別處理了，保證別重複處理相同的訊息即可。 - 比如基於資料庫的唯一鍵來保證重複資料不會重複插入多條。因為有唯一鍵約束了，重複資料插入只會報錯，不會導致資料庫中出現髒資料。  當然，如何保證 MQ 的消費是冪等性的，需要結合具體的業務來看。 ## 如何保證訊息的可靠傳輸（不丟失）？ 這個是肯定的，MQ的基本原則就是資料不能多一條，也不能少一條，不能多其實就是我們前面重複消費的問題。不能少，就是資料不能丟，像計費，扣費的一些資訊，是肯定不能丟失的。 資料的丟失問題，可能出現在生產者、MQ、消費者中，咱們從 RabbitMQ 和 Kafka 分別來分析一下吧。 ### RabbitMQ如何保證訊息的可靠  **生產者丟資料** 生產者將資料傳送到 RabbitMQ 的時候，可能資料就在半路給搞丟了，因為網路問題啥的，都有可能。 此時可以選擇用 RabbitMQ 提供的事務功能，就是生產者**傳送資料之前**開啟 RabbitMQ 事務`channel.txSelect`，然後傳送訊息，如果訊息沒有成功被 RabbitMQ 接收到，那麼生產者會收到異常報錯，此時就可以回滾事務`channel.txRollback`，然後重試傳送訊息；如果收到了訊息，那麼可以提交事務`channel.txCommit`。 ```java // 開啟事務 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 去處理，訊息是不會丟的。  ### Kakfa如何保證訊息的可靠 - **消費者丟資料** 唯一可能導致消費者弄丟資料的情況，就是說，你消費到了這個訊息，然後消費者那邊**自動提交了 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 都同步到了訊息之後，才認為本次寫成功了。如果沒滿足這個條件，生產者會自動不斷的重試，重試無限次。 ## 如何保證訊息的順序性？ 我舉個例子，我們以前做過一個 mysql `binlog` 同步的系統，壓力還是非常大的，日同步資料要達到上億，就是說資料從一個 mysql 庫原封不動地同步到另一個 mysql 庫裡面去（mysql -> mysql）。常見的一點在於說比如大資料 team，就需要同步一個 mysql 庫過來，對公司的業務系統的資料做各種複雜的操作。 你在 mysql 裡增刪改一條資料，對應出來了增刪改 3 條 `binlog` 日誌，接著這三條 `binlog` 傳送到 MQ 裡面，再消費出來依次執行，起碼得保證人家是按照順序來的吧？不然本來是：增加、修改、刪除；你楞是換了順序給執行成刪除、修改、增加，不全錯了麼。 本來這個資料同步過來，應該最後這個資料被刪除了；結果你搞錯了這個順序，最後這個資料保留下來了，資料同步就出錯了。 先看看順序會錯亂的倆場景： - **RabbitMQ**：一個 queue，多個 consumer。比如，生產者向 RabbitMQ 裡傳送了三條資料，順序依次是 data1/data2/data3，壓入的是 RabbitMQ 的一個記憶體佇列。有三個消費者分別從 MQ 中消費這三條資料中的一條，結果消費者2先執行完操作，把 data2 存入資料庫，然後是 data1/data3。這不明顯亂了。  * **Kafka**：比如說我們建了一個 topic，有三個 partition。生產者在寫的時候，其實可以指定一個 key，比如說我們指定了某個訂單 id 作為 key，那麼這個訂單相關的資料，一定會被分發到同一個 partition 中去，而且這個 partition 中的資料一定是有順序的。 消費者從 partition 中取出來資料的時候，也一定是有順序的。到這裡，順序還是 ok 的，沒有錯亂。接著，我們在消費者裡可能會搞**多個執行緒來併發處理訊息**。因為如果消費者是單執行緒消費處理，而處理比較耗時的話，比如處理一條訊息耗時幾十 ms，那麼 1 秒鐘只能處理幾十條訊息，這吞吐量太低了。而多個執行緒併發跑的話，順序可能就亂掉了。  ### RabbitMQ解決方案 拆分多個 queue，每個 queue 一個 consumer，就是多一些 queue 而已，確實是麻煩點；或者就一個 queue 但是對應一個 consumer，然後這個 consumer 內部用記憶體佇列做排隊，然後分發給底層不同的 worker 來處理。  ### Kafka解決方案 - 一個 topic，一個 partition，一個 consumer，內部單執行緒消費，單執行緒吞吐量太低，一般不會用這個。 - 寫 N 個記憶體 queue，具有相同 key 的資料都到同一個記憶體 queue；然後對於 N 個執行緒，每個執行緒分別消費一個記憶體 queue 即可，這樣就能保證順序性。  ## 如何處理訊息推積？ ### 大量訊息在 mq 裡積壓了幾個小時了還沒解決 一個消費者一秒是 1000 條，一秒 3 個消費者是 3000 條，一分鐘就是 18 萬條。所以如果你積壓了幾百萬到上千萬的資料，即使消費者恢復了，也需要大概 1 小時的時間才能恢復過來。 一般這個時候，只能臨時緊急擴容了，具體操作步驟和思路如下： - 先修復 consumer 的問題，確保其恢復消費速度，然後將現有 consumer 都停掉。 - 新建一個 topic，partition 是原來的 10 倍，臨時建立好原先 10 倍的 queue 數量。 - 然後寫一個臨時的分發資料的 consumer 程式，這個程式部署上去消費積壓的資料，**消費之後不做耗時的處理**，直接均勻輪詢寫入臨時建立好的 10 倍數量的 queue。 - 接著臨時徵用 10 倍的機器來部署 consumer，每一批 consumer 消費一個臨時 queue 的資料。這種做法相當於是臨時將 queue 資源和 consumer 資源擴大 10 倍，以正常的 10 倍速度來消費資料。 - 等快速消費完積壓資料之後，**得恢復原先部署的架構**，**重新**用原先的 consumer 機器來消費訊息。 ### mq 中的訊息過期失效了 假設你用的是 RabbitMQ，RabbtiMQ 是可以設定過期時間的，也就是 TTL。如果訊息在 queue 中積壓超過一定的時間就會被 RabbitMQ 給清理掉，這個資料就沒了。那這就是第二個坑了。這就不是說資料會大量積壓在 mq 裡，而是**大量的資料會直接搞丟**。 這個情況下，就不是說要增加 consumer 消費積壓的訊息，因為實際上沒啥積壓，而是丟了大量的訊息。我們可以採取一個方案，就是**批量重導**，這個我們之前線上也有類似的場景幹過。就是大量積壓的時候，我們當時就直接丟棄資料了，然後等過了高峰期以後，比如大家一起喝咖啡熬夜到晚上12點以後，使用者都睡覺了。這個時候我們就開始寫程式，將丟失的那批資料，寫個臨時程式，一點一點的查出來，然後重新灌入 mq 裡面去，把白天丟的資料給他補回來。也只能是這樣了。 假設 1 萬個訂單積壓在 mq 裡面，沒有處理，其中 1000 個訂單都丟了，你只能手動寫程式把那 1000 個訂單給查出來，手動發到 mq 裡去再補一次。 ### mq 都快寫滿了 如果訊息積壓在 mq 裡，你很長時間都沒有處理掉，此時導致 mq 都快寫滿了，咋辦？這個還有別的辦法嗎？沒有，誰讓你第一個方案執行的太慢了，你臨時寫程式，接入資料來消費，**消費一個丟棄一個，都不要了**，快速消費掉所有的訊息。然後走第二個方案，到了晚上再補資料吧。 參考資料： * Kafa深度解析 * RabbitMQ原始碼解析 **好了，今天就到這兒吧，我是冰河，大家有啥問題可以在下方留言，也可以加我微信：sun_shine_lyz，我拉你進群，一起交流技術，一起進階，一起