消息隊列MQ

• 使用消息隊列場景和好處
• 使用消息隊列會帶來什麼問題，有什麼解決方案
• 如何使用MQ（以ActiveMQ爲例的簡單例子）

1.消息隊列的應用場景和好處：

異步-流量削峯

　　咱們先來看下傳統的服務器接收處理請求的流程

　　如上圖，在不使用消息隊列服務器的時候，用戶的請求都直懟數據庫，在高併發的狀況下數據庫壓力劇增，不只使得響應速度變慢，還可能所以而掛掉數據庫，致使用戶頁面直接報錯，項目經理找上門，而後#！%@！#* ......（PS：儘管是某服務掛了，但某寶的用戶頁面提示信息必定會甩鍋給網絡不通哦~）

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

　　如上圖，在使用消息隊列以後，即便在高併發的狀況下用戶的請求數據發送給消息隊列以後當即返回，再由消息隊列的消費者進程從消息隊列中獲取數據，異步寫入數據庫。因爲消息隊列服務器處理消息速度比數據庫要快不少，所以響應速度（用戶體驗感）獲得大幅改善。

　　所以咱們能夠得出消息隊列具備很好的流量削峯做用的功能——即經過異步處理，將短期高併發產生的事務消息存儲在消息隊列中，從而削去高峯期的併發事務。如在某些電商平臺的一些秒殺活動中，合理使用消息隊列能夠抵禦活動剛開始大量請求涌入對系統的衝擊。　　由於用戶請求數據寫入消息隊列以後就當即返回給用戶了，可是請求數據在後續的業務校驗、寫數據庫等操做中可能失敗。所以使用消息隊列進行異步處理以後，須要適當修改業務流程進行配合，好比用戶在提交訂單以後，訂單數據寫入消息隊列，不能當即返回用戶訂單提交成功，須要在消息隊列的訂單消費者進程真正處理完該訂單以後，甚至出庫後，再經過電子郵件或短信通知用戶訂單成功，以避免交易糾紛。這就相似咱們平時手機訂火車票等。

異步-系統解耦

　我看也先來看看傳統的系統數據傳輸模式

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

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

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

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

　　另外爲了不消息隊列服務器宕機形成消息丟失，會將成功發送到消息隊列的消息存儲在消息生產者服務器上，等消息真正被消費者服務器處理後才刪除消息。在消息隊列服務器宕機後，生產者服務器會選擇分佈式消息隊列服務器集羣中的其餘服務器發佈消息。

　　 除發佈訂閱模式以外，消息隊列還有其餘的傳輸模式

　　點對點模型

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

　　生產者消費者模型

　　若是發送者和接收者均可以有多個部署實例，甚至不一樣的類型；可是共用同一個隊列，這就變成了標準的生產者消費者模型。在該模型，三個角色通常稱爲生產（Producer）、分佈式隊列（Queue）、消費者（Consumer）。

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

2.使用消息隊列會帶來什麼問題：

可用性下降： 在加入MQ以前，你不用考慮MQ服務器掛掉的狀況，引入MQ以後你就須要去考慮了，可用性下降。

複雜性提升： 加入MQ以後，你須要保證消息沒有被重複消費、處理消息丟失的狀況、保證消息傳遞的順序性等問題。所以須要考慮的東西更多，系統複雜性增大。

數據一致性：消息隊列帶來的異步確實能夠提升系統響應速度，可是，萬一消息的真正消費者並無正確消費消息怎麼辦？這樣就會致使數據不一致的狀況了。

對於可用性問題

　　引入消息隊列後，系統的可用性降低。實際項目中發送MQ消息，若是不作集羣，其中mq機器出了故障宕機了，那麼mq消息就不能發送了，系統就崩潰了，因此咱們須要集羣MQ，當其中一臺MQ出了故障，其他的MQ機器能夠接着繼續運轉，在生產中，沒人使用單機的消息隊列。若是有，那確定爲了用而用（顯得技術複雜一下，好忽悠多收點錢），對於這個問題，須要對MQ集羣技術有比較深入的理解，各類消息中間件的集羣方式不一樣，下面以ActiveMq的集羣爲例（Zookeeper+ActiveMq）,先看圖

　　這種應用場景叫作master/slave（主/備模式），在這種場景下，我有三臺服務器（主和備），任何狀況下，只有「主」在工做，「備」是在主出現故障時，接替「主」來提供服務。在zookeeper的支持下，這一過程是這樣實現的，Zookeeper提供目錄和節點的服務，當個人三臺服務器啓動時，會在zookeeper的指定目錄下建立對應本身的臨時節點（這個過程稱爲「註冊」），所謂臨時節點，是靠心跳（定時向zookeeper服務器發送數據包）維繫，當主服務器出現故障（沒法向zookeeper服務器發送數據包，zookeeper會刪除改臨時節點。服務器向zookeeper註冊時，zookeeper會分配序列號，咱們認爲序列號小的那個，就是「主」，序列號大的那個，就是「備」。

　　當咱們的客戶端（一般是web server）須要訪問服務時，須要鏈接zookeeper，得到指定目錄下的臨時節點列表，也就是已經註冊的服務器信息，得到序列號小的那臺「主」服務器的地址，進行後續的訪問操做。以達到「老是訪問主服務器」的目的。當「主」服務器發生故障，zookeeper從指定目錄下刪除對應的臨時節點，同時能夠通知關心這一變化的全部客戶端，高效且迅速的傳播這一信息。當下個請求來的時候，仍是鏈接zookeeper，可是此時實際上是訪問備用的MQ。

對於複雜性問題

　　1.如何保證消息不被重複消費呢？　　要回答好這個問題，首先要知道爲何消息會被重複消費，大多都是由於網絡不通致使，確認信息沒有傳送到消息隊列，致使消息隊列不知道本身已經消費過該消息了，再次將該消息分發給其餘的消費者。因此解決問題的方式有以下三種思路

　　1.若是消息是作數據庫的插入操做，給這個消息作一個惟一主鍵，那麼就算出現重複消費的狀況，就會致使主鍵衝突，避免數據庫出現髒數據。  2.若是你拿到這個消息作redis的set的操做，不用解決，由於你不管set幾回結果都是同樣的，set操做原本就算冪等操做。  3.若是上面兩種狀況還不行，準備一個第三服務方來作消費記錄。以redis爲例，給消息分配一個全局id，只要消費過該消息，將 以K-V形式寫入redis。那消費者開始消費前，先去redis中查詢有沒消費記錄便可。

　　2.如何保證消息的可靠性傳輸呢？

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

　　其實這些問題早在中間件開發者已經考慮到了，也提供了一些可配置的文件給咱們自行設定相關參數，消息隊列通常都會持久化到磁盤這個不用擔憂，而後生產者數據丟失的話MQ的事務會回滾，能夠嘗試從新發送，消費者丟的的話通常都是採用了自動確認消息模式致使消費信息被刪，只要修改成手動確認就好了，也就是說消費者消費完以後，調用一個MQ的確認方法就好了

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

　　經過算法，將須要保持前後順序的消息放到同一個消息隊列中，而後只用一個消費者去消費該隊列。

　　（1）rabbitmq：拆分多個queue，每一個queue一個consumer，就是多一些queue而已，確實是麻煩點；或者就一個queue可是對應一個consumer，而後這個consumer內部用內存隊列作排隊，而後分發給底層不一樣的worker來處理

　　（2）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模式實現起來會相對簡單一些，可是實時性取決於輪訓的頻率，在對實時性要求高的場景不適合使用。

如何使用MQ（以ActiveQM爲例）

　　附上官網：http://activemq.apache.org/

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

　　附上代碼，jar包本身下 https://pan.baidu.com/s/1SUBoypW-w--KeFjHsOtg

　　發佈訂閱模式

　　生產者-發佈

public class JMSProducer {

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; // 默認的鏈接地址
    private static final int SENDNUM=10; // 發送的消息數量複製代碼

public static void main(String[] args) {複製代碼

ConnectionFactory connectionFactory; // 鏈接工廠
        Connection connection = null; // 鏈接
        Session session; // 會話 接受或者發送消息的線程
        Destination destination; // 消息的目的地
        MessageProducer messageProducer; // 消息生產者複製代碼

// 實例化鏈接工廠
        connectionFactory=new ActiveMQConnectionFactory(JMSProducer.USERNAME, JMSProducer.PASSWORD, JMSProducer.BROKEURL);複製代碼

try {
            connection=connectionFactory.createConnection(); // 經過鏈接工廠獲取鏈接
            connection.start(); // 啓動鏈接
            session=connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE); // 建立Session
            // destination=session.createQueue("FirstQueue1"); // 建立消息隊列
            destination=session.createTopic("FirstTopic1");
            messageProducer=session.createProducer(destination); // 建立消息生產者
            sendMessage(session, messageProducer); // 發送消息
            session.commit();
        } catch (Exception e) {
            // TODO Auto-generated catch block
            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;i<JMSProducer.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();
        } 
    }
}複製代碼

　　end

　　by the way

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

　　

做者：卓慶森出處：https://www.cnblogs.com/zhuoqingsen/