消息隊列

一、爲何要用消息隊列

解耦、異步、削峯

 

 

 

 

A系統調用B系統、C系統，傳統的調用是直接調用，可是當B系統說我不須要你提供數據了，這時候A須要改代碼，Ｃ系統說我不須要某個字段了，這時候Ａ也要改代碼，若是又多了一個D系統，A又要寫代碼。爲了實現解耦，引入消息隊列，Ａ將產生的數據丟到消息隊列中，哪一個系統須要　哪一個系統就去取；
A系統調用B系統，B系統因爲某個須要調用第三方接口超時，致使A系統響應速度慢，而B系統的好壞又不會影響業務邏輯，因此能夠改成A異步調用B，A將消息丟到消息隊列中，B系統訂閱消息，實現A的快速響應；
當大量流量請求到系統A時，因爲數據庫的處理能力有限，形成數據庫鏈接異常。使用消息隊列，大量請求先丟到消息隊列中，系統A使用按批拉數據的方式，批量處理數據，生產中，高峯期短暫的消息積壓是容許的。

 

二、使用消息隊列有什麼缺點

系統複雜性增長：加了消息隊列，須要保證消息不會重複消費，須要保證消息的可靠性，須要保證消息隊列的高可用
系統的可用性下降：若是消息隊列掛了，那麼系統也會受到影響

 

三、RocketMQ和ActiveMQ的區別

ActiveMQ嚴格遵循JMS規範，可持久化到內存、文件、數據庫，可用性高主要是主從，多語言支持，消失丟失率低；
RocketMQ持久化到磁盤文件，可用性很是高，支持分佈式，只支持Java，消息理論上不會丟失；

 

四、MQ可否保證消息必達，即消息的可靠性（如何處理消息丟失的問題）？

 

 

 

 

 

丟數據，mq通常分爲兩種，一種是mq本身弄丟數據，一種是消費的時候弄丟數據。以rabbitmq爲例。

（1）生產者弄丟數據

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

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

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，你也是能夠本身重發的。


（3）消費端弄丟了數據

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

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

 

五、如何保證消息不被重複消費啊（如何保證消息消費時的冪等性）？

 

 

保證冪等性

（1）好比數據要寫庫時，你先根據主鍵查一下，若是這數據都存在了，就無需插入，update一下就能夠了

（2）好比寫redis，每次都是set，自然冪等性

（3）好比你不是上面兩個場景，那作的稍微複雜一點，你須要讓生產者發送每條數據的時候，裏面加一個全局惟一的id，相似訂單id之類的東西，而後你這裏消費到了以後，先根據這個id去好比redis裏查一下，以前消費過嗎？若是沒有消費過，你就處理，而後這個id寫redis。若是消費過了，那你就別處理了，保證別重複處理相同的消息便可。

（4）好比基於數據庫的惟一鍵來保證重複數據不會重複插入多條，咱們以前線上系統就有這個問題，就是拿到數據的時候，每次重啓可能會有重複，由於kafka消費者還沒來得及提交offset，重複數據拿到了之後咱們插入的時候，由於有惟一鍵約束了，因此重複數據只會插入報錯，不會致使數據庫中出現髒數據

 

六、如何保證消息的順序性？

（1）rabbitmq保證數據的順序性若是存在多個消費者，那麼就讓每一個消費者對應一個queue，而後把要發送 的數據全都放到一個queue，這樣就能保證全部的數據只到達一個消費者從而保證每一個數據到達數據庫都是順序的。rabbitmq：拆分多個queue，每一個queue一個consumer，就是多一些queue而已，確實是麻煩點；或者就一個queue可是對應一個consumer，而後這個consumer內部用內存隊列作排隊，而後分發給底層不一樣的worker來處理。

 

 

（2）kafka保證數據的順序性
 kafka 寫入partion時指定一個key，列如訂單id，那麼消費者從partion中取出數據的時候確定是有序的，當開啓多個線程的時候可能致使數據不一致，這時候就須要內存隊列，將相同的hash過的數據放在一個內存隊列裏，這樣就能保證一條線程對應一個內存隊列的數據寫入數據庫的時候順序性的，從而能夠開啓多條線程對應多個內存隊列kafka：一個topic，一個partition，一個consumer，內部單線程消費，寫N個內存queue，而後N個線程分別消費一個內存queue便可

 

七、如何解決消息隊列的延時以及過時失效問題？消息隊列滿了之後該怎麼處理？有幾百萬消息持續積壓幾小時，說說怎麼解決？

通常這個時候，只能操做臨時緊急擴容了，具體操做步驟和思路以下：
（1）先修復consumer的問題，確保其恢復消費速度，而後將現有cnosumer都停掉

（2）新建一個topic，partition是原來的10倍，臨時創建好原先10倍或者20倍的queue數量

（3）而後寫一個臨時的分發數據的consumer程序，這個程序部署上去消費積壓的數據，消費以後不作耗時的處理，直接均勻輪詢寫入臨時創建好的10倍數量的queue

（4）接着臨時徵用10倍的機器來部署consumer，每一批consumer消費一個臨時queue的數據

（5）這種作法至關因而臨時將queue資源和consumer資源擴大10倍，以正常的10倍速度來消費數據

（6）等快速消費完積壓數據以後，得恢復原先部署架構，從新用原先的consumer機器來消費消息