關於MQ那些事_ZenDei技術網路在線

我走過最長的路是你的套路女：二號男嘉賓，假如我們牽手成功後，你會買名牌包包給我嗎？男：那你會聽話嗎？女：會聽話。男：聽話咱不買！那麼消息隊列MQ有什麼套路呢？（這個話題轉換生硬度連我自己都怕！） 1.消息隊列的應用場景和好處：非同步-流量削峰非同步-流量削峰我們先來看下傳統的伺服器接 ...

我走過最長的路是你的套路

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

男：那你會聽話嗎？

女：會聽話。

男：聽話咱不買！

那麼消息隊列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）,先看圖

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

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

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

對於複雜性問題

如何保證消息不被重覆消費呢？

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

　　1.如果消息是做資料庫的插入操作，給這個消息做一個唯一主鍵，那麼就算出現重覆消費的情況，就會導致主鍵衝突，避免資料庫出現臟數據。
2.如果你拿到這個消息做redis的set的操作，不用解決，因為你無論set幾次結果都是一樣的，set操作本來就算冪等操作。
3.如果上面兩種情況還不行，準備一個第三服務方來做消費記錄。以redis為例，給消息分配一個全局id，只要消費過該消息，將<id,message>以K-V形式寫入redis。那消費者開始消費前，先去redis中查詢有沒消費記錄即可。

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

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

　　其實這些問題早在中間件開發者已經考慮到了，也提供了一些可配置的文件給我們自行設定相關參數，消息隊列一般都會持久化到磁碟這個不用擔心，然後生產者數據丟失的話可以嘗試重新發送，消費者丟的的話一般都是採用了自動確認消息模式導致消費信息被刪，只要修改為手動確認就行了

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

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

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

　　　　1.定期刪除過期的數據

　　　　2.RejectNewEvents 拒絕新來的消息，向Producer返回RejectNewEvents錯誤碼。

　　　　3.按照特定策略丟棄已有消息

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

Push模型實時性好，但是因為狀態維護等問題，難以應用到消息中間件的實踐中。
Pull模式實現起來會相對簡單一些，但是實時性取決於輪訓的頻率，在對實時性要求高的場景不適合使用。

3.如何使用MQ（以ActiveQM為例）

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

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

　　附上代碼，jar包自己下 https://pan.baidu.com/s/1SUBoypW-w_--KeFj_HsOtg

　　發佈訂閱模式

　　生產者-發佈

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

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

　　轉載請註明出處，謝謝

關於MQ那些事

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

非同步-流量削峰

非同步-系統解耦