1. 消息隊列的應用場景和好處：2.使用消息隊列會帶來什麼問題：2.1有什麼解決方案2.如何保證消息的可靠性傳輸呢？3.如何保證從消息隊列裡拿到的數據按順序執行？3.如何使用MQ（以ActiveQM為例） 我走過最長的路是你的套路

　　女：二号男嘉賓，假如我們牽手成功後，你會買名牌包包給我嗎？

　　男：那你會聽話嗎？

　　女：會 聽話。

　　男：聽話 咱不買！

　　OK那麼消息隊列MQ有什麼套路呢？（這個話題轉換生硬度連我自己都怕！）

　　使用消息隊列場景和好處使用消息隊列會帶來什麼問題，有什麼解決方案如何使用MQ（以ActiveMQ為例的簡單例子）1. 消息隊列的應用場景和好處：異步-流量削峰 我們先來看下傳統的服務器接收處理請求的流程

　　如上圖，在不使用消息隊列服務器的時候，用戶的請求都直怼數據庫，在高并發的情況下數據庫壓力劇增，不僅使得響應速度變慢，還可能因此而挂掉數據庫，導緻用戶頁面直接報錯，項目經理找上門，然後*#！%@！#** ......（PS：盡管是某服務挂了，但某寶的用戶頁面提示信息一定會甩鍋給網絡不通哦~）

　　我們再來看加入消息隊列服務器之後的接收處理請求的流程會發生什麼變化

　　如上圖，在使用消息隊列之後，即使在高并發的情況下用戶的請求數據發送給消息隊列之後立即返回，再由消息隊列的消費者進程從消息隊列中獲取數據，異步寫入數據庫。由于消息隊列服務器處理消息速度比數據庫要快很多，因此響應速度（用戶體驗感）得到大幅改善。

　　因此我們可以得出消息隊列具有很好的流量削峰作用的功能——即通過異步處理，将短時間高并發産生的事務消息存儲在消息隊列中，從而削去高峰期的并發事務。如在某些電商平台的一些秒殺活動中，合理使用消息隊列可以抵禦活動剛開始大量請求湧入對系統的沖擊。

　　因為用戶請求數據寫入消息隊列之後就立即返回給用戶了，但是請求數據在後續的業務校驗、寫數據庫等操作中可能失敗。因此使用消息隊列進行異步處理之後，需要适當修改業務流程進行配合，比如用戶在提交訂單之後，訂單數據寫入消息隊列，不能立即返回用戶訂單提交成功，需要在消息隊列的訂單消費者進程真正處理完該訂單之後，甚至出庫後，再通過電子郵件或短信通知用戶訂單成功，以免交易糾紛。這就類似我們平時手機訂火車票等。

　　異步-系統解耦 我看也先來看看傳統的系統數據傳輸模式

　　如上圖，主系統和其他系統的耦合性太強，都是直接調用，稍微有一點改動或者新增模塊，雙方都得改代碼，過于麻煩

　　然後，我們再來看看加入了消息隊列之後，系統的結構會發生什麼變化

　　如上圖，我們知道如果模塊之間不存在直接調用，那麼新增模塊或者修改模塊就對其他模塊影響較小，這樣系統的可擴展性無疑更好一些。

　　消息隊列是利用發布-訂閱模式工作，消息發送者（生産者）發布消息，一個或多個消息接受者（消費者）訂閱消息。從上圖可以看到消息發送者（生産者）和消息接受者（消費者）之間沒有直接耦合，消息發送者将消息發送至分布式消息隊列即結束對消息的處理，消息接受者從分布式消息隊列獲取該消息後進行後續處理，并不需要知道該消息從何而來。對新增業務，隻要對該類消息感興趣，即可訂閱該消息，對原有系統和業務沒有任何影響，從而實現網站業務的可擴展性設計。

　　另外為了避免消息隊列服務器宕機造成消息丢失，會将成功發送到消息隊列的消息存儲在消息生産者服務器上，等消息真正被消費者服務器處理後才删除消息。在消息隊列服務器宕機後，生産者服務器會選擇分布式消息隊列服務器集群中的其他服務器發布消息。

　　除發布訂閱模式之外，消息隊列還有其他的傳輸模式

　　點對點模型

　　基礎模型中，隻有一個發送者、一個接收者和一個分布式隊列。

　　生産者消費者模型

　　如果發送者和接收者都可以有多個部署實例，甚至不同的類型；但是共用同一個隊列，這就變成了标準的生産者消費者模型。在該模型，三個角色一般稱為生産（Producer）、分布式隊列（Queue）、消費者（Consumer）。

　　中途小結：消息隊列對系統的并發處理的能力和擴展性有所提升

　　2.使用消息隊列會帶來什麼問題：可用性降低：在加入MQ之前，你不用考慮MQ服務器挂掉的情況，引入MQ之後你就需要去考慮了，可用性降低。複雜性提高：加入MQ之後，你需要保證消息沒有被重複消費、處理消息丢失的情況、保證消息傳遞的順序性等問題。因此需要考慮的東西更多，系統複雜性增大。數據一緻性：消息隊列帶來的異步确實可以提高系統響應速度，但是，萬一消息的真正消費者并沒有正确消費消息怎麼辦？這樣就會導緻數據不一緻的情況了。 2.1有什麼解決方案

　　對于可用性問題 引入消息隊列後，系統的可用性下降。實際項目中發送MQ消息，如果不做集群，其中mq機器出了故障宕機了，那麼mq消息就不能發送了，系統就崩潰了，所以我們需要集群MQ，當其中一台MQ出了故障，其餘的MQ機器可以接着繼續運轉，在生産中，沒人使用單機的消息隊列。如果有，那肯定為了用而用（顯得技術複雜一下，好忽悠多收點錢），對于這個問題，需要對MQ集群技術有比較深刻的理解，各種消息中間件的集群方式不同，下面以ActiveMq的集群為例（zookeeper ActiveMq）,先看圖

　　eeper會删除改臨時節點。服務器向zookeeper注冊時，zookeeper會分配序列号，我們認為序列号小的那個，就是“主”，序列号大的那個，就是“備”。

　　當我們的客戶端（通常是web server）需要訪問服務時，需要連接zookeeper，獲得指定目錄下的臨時節點列表，也就是已經注冊的服務器信息，獲得序列号小的那台“主”服務器的地址，進行後續的訪問操作。以達到“總是訪問主服務器”的目的。當“主”服務器發生故障，zookeeper從指定目錄下删除對應的臨時節點，同時可以通知關心這一變化的所有客戶端，高效且迅速的傳播這一信息。當下個請求來的時候，還是連接zookeeper，但是此時其實是訪問備用的MQ。

　　對于如何配置集群，這裡就不演示，自行網上搜教程，一大把的！

　　對于複雜性問題 1.如何保證消息不被重複消費呢？

　　要回答好這個問題，首先要知道為什麼消息會被重複消費，大多都是因為網絡不通導緻，确認信息沒有傳送到消息隊列，導緻消息隊列不知道自己已經消費過該消息了，再次将該消息分發給其他的消費者。所以解決問題的方式有如下三種思路

　　如果消息是做數據庫的插入操作，給這個消息做一個唯一主鍵，那麼就算出現重複消費的情況，就會導緻主鍵沖突，避免數據庫出現髒數據。如果你拿到這個消息做redis的set的操作，不用解決，因為你無論set幾次結果都是一樣的，set操作本來就算幂等操作。如果上面兩種情況還不行，準備一個第三服務方來做消費記錄。以redis為例，給消息分配一個全局id，隻要消費過該消息，将id,message以K-V形式寫入redis。那消費者開始消費前，先去redis中查詢有沒消費記錄即可。 2.如何保證消息的可靠性傳輸呢？

　　其實這個問題是第一個問題的擴展，換而言之，我們要保證可靠性傳輸，其實就是保證防止生産者弄丢數據、消息隊列弄丢數據、消費者弄丢數據而已

　　其實這些問題早在中間件開發者已經考慮到了，也提供了一些可配置的文件給我們自行設定相關參數，消息隊列一般都會持久化到磁盤這個不用擔心，然後生産者數據丢失的話MQ的事務會回滾，可以嘗試重新發送，消費者丢的的話一般都是采用了自動确認消息模式導緻消費信息被删，隻要修改為手動确認就行了，也就是說消費者消費完之後，調用一個MQ的确認方法就行了

　　3.如何保證從消息隊列裡拿到的數據按順序執行？

　　通過算法，将需要保持先後順序的消息放到同一個消息隊列中，然後隻用一個消費者去消費該隊列。

　　rabbitmq：拆分多個queue，每個queue一個consumer，就是多一些queue而已，确實是麻煩點；或者就一個queue但是對應一個consumer，然後這個consumer内部用内存隊列做排隊，然後分發給底層不同的worker來處理kafka：一個topic，一個partition，一個consumer，内部單線程消費，寫N個内存queue，然後N個線程分别消費一個内存queue即可 4.如何解決消息隊列的延時以及過期失效問題？有幾百萬消息持續積壓幾小時，怎麼解決？

　　這個問題是生産環境出現事故後的，考察你如何快速的解決問題，，消息隊列的延遲和過期失效是消息隊列的自我保護機制，目的是為了防止本身被擠爆，當然是可以關閉保護，比如當某個消息消費失敗5次後，就把這個消息丢棄等，盡量不要關掉保護機制，那麼問題來了，那些被丢棄的消息難道就不要了嗎？其實并不是，我們可以針對該業務，查詢出來将丢失的那批數據，寫個臨時程序，一點一點的查出來，然後重新灌入mq裡面去，把丢的數據給他補回來。

　　5.數據是通過push還是pull方式給到消費端，各自有什麼弊端？

　　Push模型實時性好，但是因為狀态維護等問題，難以應用到消息中間件的實踐中,因為在Broker端需要維護Consumer的狀态，不好适用于Broker去支持大量的Consumer的場景Consumer的消費速度是不一緻的，Broker進行推送難以處理不同的Consumer的狀況Broker難以應對Consumer無法消費消息的情況，因為不知道Consumer的宕機是短暫的還是永久的）另外推送消息（量可能會很大）也會加重Consumer的負載或者壓垮Consumer。如果對應隻有1個Consumer，用push比pull好。Pull模式實現起來會相對簡單一些，但是實時性取決于輪訓的頻率，在對實時性要求高的場景不适合使用。3.如何使用MQ（以ActiveQM為例） 附上官網：http://activemq.apache.org/

　　附上啟動服務訪問地址：http://127.0.0.1:8161/admin/ 用戶名/密碼 admin/admin

　　附上代碼，jar包自己下 httblock e.printStackTrace(); } finally{ if(connection!=null){ try { connection.close(); } catch (JMSException e) { // TODO Auto-generated catch block e.printStackTrace(); } } } } /** * 發送消息 * @param session * @param messageProducer * @throws Exception */ public static void sendMessage(Session session,MessageProducer messageProducer)throws Exception{ for(int i=0;iJMSProducer.SENDNUM;i ){ TextMessage message=session.createTextMessage("ActiveMQ 發送的消息" i); System.out.println("發送消息：" "ActiveMQ 發布的消息" i); messageProducer.send(message); } } }

　　消費者-訂閱

　　/** * 消息監聽-訂閱者一 * @author Administrator * */ public class Listener implements MessageListener{ @Override public void onMessage(Message message) { // TODO Auto-generated method stub try { System.out.println("訂閱者一收到的消息：" ((TextMessage)message).getText()); } catch (JMSException e) { // TODO Auto-generated catch block e.printStackTrace(); } } } public class JMSConsumer { private static final String USERNAME=ActiveMQConnection.DEFAULT_USER; // 默認的連接用戶名 private static final String PASSWORD=ActiveMQConnection.DEFAULT_PASSWORD; // 默認的連接密碼 private static final String BROKEURL=ActiveMQConnection.DEFAULT_BROKER_URL; // 默認的連接地址 public static void main(String[] args) { ConnectionFactory connectionFactory; // 連接工廠 Connection connection = null; // 連接 Session session; // 會話 接受或者發送消息的線程 Destination destination; // 消息的目的地 MessageConsumer messageConsumer; // 消息的消費者 // 實例化連接工廠 connectionFactory=new ActiveMQConnectionFactory(JMSConsumer.USERNAME, JMSConsumer.PASSWORD, JMSConsumer.BROKEURL); try { connection=connectionFactory.createConnection(); // 通過連接工廠獲取連接 connection.start(); // 啟動連接 session=connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE); // 創建Session // destination=session.createQueue("FirstQueue1"); // 創建連接的消息隊列 destination=session.createTopic("FirstTopic1"); messageConsumer=session.createConsumer(destination); // 創建消息消費者 messageConsumer.setMessageListener(new Listener()); // 注冊消息監聽 } catch (JMSException e) { // TODO Auto-generated catch block e.printStackTrace(); } } }

　　by the way

　　我認為一個優秀的分布式消息隊列，應該具備以下的能力：高吞吐、低時延（因場景而異），傳輸透明，伸縮性強，有冗災能力，一緻性順序投遞，同步 異步的發送方式，完善的運維和監控工具和開源。

　　來源：http://t.cn/AilfqhYu

　　:-D 搜索微信号（ID：芋道源碼），可以獲得各種 Java 源碼解析、原理講解、面試題、學習指南。

　　:-D 并且，回複【書籍】後，可以領取筆者推薦的各種 Java 從入門到架構的 100 本書籍。

　　:-D 并且，回複【技術群】後，可以加入專門讨論 Java、後端、架構的技術群。

　　來吧，騷年~

　　,

更多精彩资讯请关注tft每日頭條，我们将持续为您更新最新资讯!