訊息佇列詳解
轉載：https://blog.csdn.net/qq_36236890/article/details/81174504
說明：此文是筆者對中華石衫老師對訊息佇列講解的一篇總結包括筆者自己的一些理解
一、為什麼使用訊息佇列？
訊息佇列使用的場景和中介軟體有很多，但解決的核心問題主要是：非同步、解耦、消峰填谷。
二、訊息佇列的優缺點
非同步、解耦、消峰填谷這是訊息佇列最大的優點，除了這些訊息佇列還可以會解決一些我們特殊業務場景的問題。但是缺點主要在於系統的可用性、複雜性、一致性問題，引入訊息佇列後，需要考慮MQ的可用性，萬一MQ崩潰了豈不是要爆炸？而且複雜性明顯提高了，需要考慮一些訊息佇列的常見問題和解決方案，還有就是一致性問題，一條訊息由多個消費者消費，萬一有一個消費者消費失敗了，就會導致資料不一致。
三、訊息佇列選型
目前常見和使用廣泛的MQ有ActiveMQ、RabbitMQ、RocketMQ、Kakfa,其特性如下：
個人總結：
ActiveMQ早期用的比較多，但是現在貌似用的都不是很多了，網上也沒有大規模吞吐量的使用案例分析，社群也貌似不是很活躍了，如果是新專案不建議採用ActiveMQ。
RabbitMQ現在使用的較為多一些，社群活躍度也很高，功能也很強大，官方還提供了管理的web介面，效能也很好，但是RabbitMQ效能好的主要原因是因為使用erlang語言開發的，erlang語言貌似天生效能好，但對於我們java開發者來說，原始碼基本看不懂，更別提深入的研究了，不過spring推出了rabbit的支援，貌似還比較好用，比自己去封裝實現並且去處理一些問題的要好多了。
RocketMQ現在開始用的人也比較多，很多人對於RocketMQ的看法是集成了Kafka和RabbitMQ的有點，是阿里開源的產品，貌似現在是捐贈給了Apache，其原始碼是java寫的，功能十分強大並且是經過阿里大規模應用的，能經過阿里實踐使用的一般來說可靠性和可用性都是相當高的，但是也存在一些小問題，現在RocketMQ雖然使用的人好像越來越多了，但是文件資料還是比較少，含金量不怎麼高，並且阿里開源的有不維護的風險，就像dubbo中間也用2年沒維護，有實力的團隊應該沒有什麼問題，小公司小團隊需要考慮一下使用RocketMQ。
Kafka就不多說了，Kafka可以說是業內標準，基本上大資料領域的實時計算、日誌、資料處理都是用kafka，開源社群異常活躍，而且像現在阿里雲、騰訊雲都推出了Kafka的雲服務，所以說Kafka就不說了，絕對沒問題，放心大膽的用吧。
最後給一個個人選型意見（不一定對啊），如果是小公司小團隊最好採用Kafka和RabbitMQ,有實力的團隊可以去搞一搞RocketMQ。
四、如何保證訊息佇列的高可用性
由於筆者只使用和實踐過RabbitMQ和Kafka，RocketMQ和ActiveMQ瞭解的不深，所以分析一下RabbitMQ和Kafka的高可用。
（一）RabbitMQ
RabbitMQ有三種模式：單機模式，普通叢集模式，映象叢集模式
（1）單機模式
單機模式平常使用在開發或者本地測試場景，一般就是測試是不是能夠正確的處理訊息，生產上基本沒人去用單機模式，風險很大。
（2）普通叢集模式
普通叢集模式就是啟動多個RabbitMQ例項。在你建立的queue，只會放在一個rabbtimq例項上，但是每個例項都同步queue的元資料。在消費的時候完了，上如果連線到了另外一個例項，那麼那個例項會從queue所在例項上拉取資料過來。
這種方式確實很麻煩，也不怎麼好，沒做到所謂的分散式，就是個普通叢集。因為這導致你要麼消費者每次隨機連線一個例項然後拉取資料，要麼固定連線那個queue所在例項消費資料，前者有資料拉取的開銷，後者導致單例項效能瓶頸。
而且如果那個放queue的例項宕機了，會導致接下來其他例項就無法從那個例項拉取，如果你開啟了訊息持久化，讓RabbitMQ落地儲存訊息的話，訊息不一定會丟，得等這個例項恢復了，然後才可以繼續從這個queue拉取資料。
這方案主要是提高吞吐量的，就是說讓叢集中多個節點來服務某個queue的讀寫操作。
（3）映象叢集模式
映象叢集模式是所謂的RabbitMQ的高可用模式，跟普通叢集模式不一樣的是，你建立的queue，無論元資料還是queue裡的訊息都會存在於多個例項上，然後每次你寫訊息到queue的時候，都會自動把訊息到多個例項的queue裡進行訊息同步。
優點在於你任何一個例項宕機了，沒事兒，別的例項都可以用。缺點在於效能開銷太大和擴充套件性很低，同步所有例項，這會導致網路頻寬和壓力很重，而且擴充套件性很低，每增加一個例項都會去包含已有的queue的所有資料，並沒有辦法線性擴充套件queue。
開啟映象叢集模式可以去RabbitMQ的管理控制檯去增加一個策略，指定要求資料同步到所有節點的，也可以要求就同步到指定數量的節點，然後你再次建立queue的時候，應用這個策略，就會自動將資料同步到其他的節點上去了。
（二）Kafka
Kafka天生就是一個分散式的訊息佇列，它可以由多個broker組成，每個broker是一個節點；你建立一個topic，這個topic可以劃分為多個partition，每個partition可以存在於不同的broker上，每個partition就放一部分資料。
kafka 0.8以前，是沒有HA機制的，就是任何一個broker宕機了，那個broker上的partition就廢了，沒法寫也沒法讀，沒有什麼高可用性可言。
kafka 0.8以後，提供了HA機制，就是replica副本機制。kafka會均勻的將一個partition的所有replica分佈在不同的機器上，來提高容錯性。每個partition的資料都會同步到吉他機器上，形成自己的多個replica副本。然後所有replica會選舉一個leader出來，那麼生產和消費都去leader，其他replica就是follower，leader會同步資料給follower。當leader掛了會自動去找replica，然後會再選舉一個leader出來，這樣就具有高可用性了。
寫資料的時候，生產者就寫leader，然後leader將資料落地寫本地磁碟，接著其他follower自己主動從leader來pull資料。一旦所有follower同步好資料了，就會發送ack給leader，leader收到所有follower的ack之後，就會返回寫成功的訊息給生產者。（當然，這只是其中一種模式，還可以適當調整這個行為）
消費的時候，只會從leader去讀，但是隻有一個訊息已經被所有follower都同步成功返回ack的時候，這個訊息才會被消費者讀到。
五、如何保證訊息消費時的冪等性
其實訊息重複消費的主要原因在於回饋機制（RabbitMQ是ack，Kafka是offset)，在某些場景中我們採用的回饋機制不同，原因也不同，例如消費者消費完訊息後回覆ack, 但是剛消費完還沒來得及提交系統就重啟了，這時候上來就pull訊息的時候由於沒有提交ack或者offset，消費的還是上條訊息。
那麼如何怎麼來保證訊息消費的冪等性呢？實際上我們只要保證多條相同的資料過來的時候只處理一條或者說多條處理和處理一條造成的結果相同即可，但是具體怎麼做要根據業務需求來定，例如入庫訊息，先查一下訊息是否已經入庫啊或者說搞個唯一約束啊什麼的，還有一些是天生保證冪等性就根本不用去管，例如redis就是天然冪等性。
還有一個問題，消費者消費訊息的時候在某些場景下要放過消費不了的訊息，遇到消費不了的訊息通過日誌記錄一下或者搞個什麼措施以後再來處理，但是一定要放過訊息，因為在某些場景下例如spring-rabbitmq的預設回饋策略是出現異常就沒有提交ack，導致了一直在重發那條消費異常的訊息，而且一直還消費不了，這就尷尬了，後果你會懂的。
六、如何保證訊息的可靠性傳輸？
由於筆者只使用和實踐過RabbitMQ和Kafka，RocketMQ和ActiveMQ瞭解的不深，所以分析一下RabbitMQ和Kafka的訊息可靠性傳輸的問題。、
（一）RabbitMQ
（1）生產者弄丟了資料
生產者將資料傳送到RabbitMQ的時候，可能資料就在半路給搞丟了，因為網路啥的問題，都有可能。此時可以選擇用RabbitMQ提供的事務功能，就是生產者傳送資料之前開啟RabbitMQ事務（channel.txSelect），然後傳送訊息，如果訊息沒有成功被RabbitMQ接收到，那麼生產者會收到異常報錯，此時就可以回滾事務（channel.txRollback），然後重試傳送訊息；如果收到了訊息，那麼可以提交事務（channel.txCommit）。但是問題是，RabbitMQ事務機制一搞，基本上吞吐量會下來，因為太耗效能。
所以一般來說，如果你要確保說寫RabbitMQ的訊息別丟，可以開啟confirm模式，在生產者那裡設定開啟confirm模式之後，你每次寫的訊息都會分配一個唯一的id，然後如果寫入了RabbitMQ中，RabbitMQ會給你回傳一個ack訊息，告訴你說這個訊息ok了。如果RabbitMQ沒能處理這個訊息，會回撥你一個nack介面，告訴你這個訊息接收失敗，你可以重試。而且你可以結合這個機制自己在記憶體裡維護每個訊息id的狀態，如果超過一定時間還沒接收到這個訊息的回撥，那麼你可以重發。
事務機制和cnofirm機制最大的不同在於，事務機制是同步的，你提交一個事務之後會阻塞在那兒，但是confirm機制是非同步的，你傳送個訊息之後就可以傳送下一個訊息，然後那個訊息RabbitMQ接收了之後會非同步回撥你一個介面通知你這個訊息接收到了。
所以一般在生產者這塊避免資料丟失，都是用confirm機制的。
（2）RabbitMQ弄丟了資料
就是RabbitMQ自己弄丟了資料，這個你必須開啟RabbitMQ的持久化，就是訊息寫入之後會持久化到磁碟，哪怕是RabbitMQ自己掛了，恢復之後會自動讀取之前儲存的資料，一般資料不會丟。除非極其罕見的是，RabbitMQ還沒持久化，自己就掛了，可能導致少量資料會丟失的，但是這個概率較小。
設定持久化有兩個步驟，第一個是建立queue的時候將其設定為持久化的，這樣就可以保證RabbitMQ持久化queue的元資料，但是不會持久化queue裡的資料；第二個是傳送訊息的時候將訊息的deliveryMode設定為2，就是將訊息設定為持久化的，此時RabbitMQ就會將訊息持久化到磁碟上去。必須要同時設定這兩個持久化才行，RabbitMQ哪怕是掛了，再次重啟，也會從磁碟上重啟恢復queue，恢復這個queue裡的資料。
而且持久化可以跟生產者那邊的confirm機制配合起來，只有訊息被持久化到磁碟之後，才會通知生產者ack了，所以哪怕是在持久化到磁碟之前，RabbitMQ掛了，資料丟了，生產者收不到ack，你也是可以自己重發的。
哪怕是你給RabbitMQ開啟了持久化機制，也有一種可能，就是這個訊息寫到了RabbitMQ中，但是還沒來得及持久化到磁碟上，結果不巧，此時RabbitMQ掛了，就會導致記憶體裡的一點點資料會丟失。
（3）消費端弄丟了資料
RabbitMQ如果丟失了資料，主要是因為你消費的時候，剛消費到，還沒處理，結果程序掛了，比如重啟了，那麼就尷尬了，RabbitMQ認為你都消費了，這資料就丟了。
這個時候得用RabbitMQ提供的ack機制，簡單來說，就是你關閉RabbitMQ自動ack，可以通過一個api來呼叫就行，然後每次你自己程式碼裡確保處理完的時候，再程式裡ack一把。這樣的話，如果你還沒處理完，不就沒有ack？那RabbitMQ就認為你還沒處理完，這個時候RabbitMQ會把這個消費分配給別的consumer去處理，訊息是不會丟的。
（二）Kafka
（1）消費端弄丟了資料
唯一可能導致消費者弄丟資料的情況，就是說，你那個消費到了這個訊息，然後消費者那邊自動提交了offset，讓kafka以為你已經消費好了這個訊息，其實你剛準備處理這個訊息，你還沒處理，你自己就掛了，此時這條訊息就丟咯。
大家都知道kafka會自動提交offset，那麼只要關閉自動提交offset，在處理完之後自己手動提交offset，就可以保證資料不會丟。但是此時確實還是會重複消費，比如你剛處理完，還沒提交offset，結果自己掛了，此時肯定會重複消費一次，自己保證冪等性就好了。
生產環境碰到的一個問題，就是說我們的kafka消費者消費到了資料之後是寫到一個記憶體的queue裡先緩衝一下，結果有的時候，你剛把訊息寫入記憶體queue，然後消費者會自動提交offset。
然後此時我們重啟了系統，就會導致記憶體queue裡還沒來得及處理的資料就丟失了
（2）kafka弄丟了資料
這塊比較常見的一個場景，就是kafka某個broker宕機，然後重新選舉partiton的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（很大很大很大的一個值，無限次重試的意思）：這個是要求一旦寫入失敗，就無限重試，卡在這裡了。
（3）生產者會不會弄丟資料
如果按照上述的思路設定了ack=all，一定不會丟，要求是，你的leader接收到訊息，所有的follower都同步到了訊息之後，才認為本次寫成功了。如果沒滿足這個條件，生產者會自動不斷的重試，重試無限次。
六、如何保證訊息的順序性
因為在某些情況下我們扔進MQ中的訊息是要嚴格保證順序的，尤其涉及到訂單什麼的業務需求，消費的時候也是要嚴格保證順序，不然會出大問題的。
先看看順序會錯亂的倆場景
- rabbitmq：一個queue，多個consumer，這不明顯亂了
- kafka：一個topic，一個partition，一個consumer，內部多執行緒，這不也明顯亂了
如何來保證訊息的順序性呢？ - rabbitmq：拆分多個queue，每個queue一個consumer，就是多一些queue而已，確實是麻煩點；或者就一個queue但是對應一個consumer，然後這個consumer內部用記憶體佇列做排隊，然後分發給底層不同的worker來處理。
- kafka：一個topic，一個partition，一個consumer，內部單執行緒消費，寫N個記憶體queue，然後N個執行緒分別消費一個記憶體queue即可。
七、如何解決訊息佇列的延時以及過期失效問題？訊息佇列滿了以後該怎麼處理？有幾百萬訊息持續積壓幾小時怎麼解決？
（一）、大量訊息在mq裡積壓了幾個小時了還沒解決
幾千萬條資料在MQ裡積壓了七八個小時，從下午4點多，積壓到了晚上很晚，10點多，11點多
這個是我們真實遇到過的一個場景，確實是線上故障了，這個時候要不然就是修復consumer的問題，讓他恢復消費速度，然後傻傻的等待幾個小時消費完畢。這個肯定不能在面試的時候說吧。
一個消費者一秒是1000條，一秒3個消費者是3000條，一分鐘是18萬條，1000多萬條，所以如果你積壓了幾百萬到上千萬的資料，即使消費者恢復了，也需要大概1小時的時間才能恢復過來。
一般這個時候，只能操作臨時緊急擴容了，具體操作步驟和思路如下：
- 先修復consumer的問題，確保其恢復消費速度，然後將現有cnosumer都停掉。
- 新建一個topic，partition是原來的10倍，臨時建立好原先10倍或者20倍的queue數量。
- 然後寫一個臨時的分發資料的consumer程式，這個程式部署上去消費積壓的資料，消費之後不做耗時的處理，直接均勻輪詢寫入臨時建立好的10倍數量的queue。
- 接著臨時徵用10倍的機器來部署consumer，每一批consumer消費一個臨時queue的資料。
- 這種做法相當於是臨時將queue資源和consumer資源擴大10倍，以正常的10倍速度來消費資料。
- 等快速消費完積壓資料之後，得恢復原先部署架構，重新用原先的consumer機器來消費訊息。
（二）、訊息佇列過期失效問題
假設你用的是rabbitmq，rabbitmq是可以設定過期時間的，就是TTL，如果訊息在queue中積壓超過一定的時間就會被rabbitmq給清理掉，這個資料就沒了。那這就是第二個坑了。這就不是說資料會大量積壓在mq裡，而是大量的資料會直接搞丟。
這個情況下，就不是說要增加consumer消費積壓的訊息，因為實際上沒啥積壓，而是丟了大量的訊息。我們可以採取一個方案，就是批量重導，這個我們之前線上也有類似的場景幹過。就是大量積壓的時候，我們當時就直接丟棄資料了，然後等過了高峰期以後，比如大家一起喝咖啡熬夜到晚上12點以後，使用者都睡覺了。
這個時候我們就開始寫程式，將丟失的那批資料，寫個臨時程式，一點一點的查出來，然後重新灌入mq裡面去，把白天丟的資料給他補回來。也只能是這樣了。
假設1萬個訂單積壓在mq裡面，沒有處理，其中1000個訂單都丟了，你只能手動寫程式把那1000個訂單給查出來，手動發到mq裡去再補一次。
(三)、訊息佇列滿了怎麼搞？
如果走的方式是訊息積壓在mq裡，那麼如果你很長時間都沒處理掉，此時導致mq都快寫滿了，咋辦？這個還有別的辦法嗎？沒有，誰讓你第一個方案執行的太慢了，你臨時寫程式，接入資料來消費，消費一個丟棄一個，都不要了，快速消費掉所有的訊息。然後走第二個方案，到了晚上再補資料吧。
個人部落格地址：http://xuyangyang.club(點選開啟)
微信訂閱號:
</div>