我走過最長的路是你的套路
女：二号男嘉賓，假如我們牽手成功後，你會買名牌包包給我嗎？
男：那你會聽話嗎？
女：會 聽話。
男：聽話 咱不買！
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）,先看圖
zookeeper會删除改臨時節點。伺服器向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為例）
不知道如何選擇的看這篇文章：ActiveMQ、RabbitMQ、RocketMQ、ZeroMQ、Kafka如何選型？寫得非常好。
附上官網：
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);
}
}
} 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
我認為一個優秀的分布式消息隊列，應該具備以下的能力：高吞吐、低延遲時間（因場景而異），傳輸透明，伸縮性強，有冗災能力，一緻性順序投遞，同步+異步的發送方式，完善的運維和監控工具和開源。
推薦去我的部落格閱讀更多：
1.Java JVM、集合、多線程、新特性系列教程
2.Spring MVC、Spring Boot、Spring Cloud 系列教程
3.Maven、Git、Eclipse、Intellij IDEA 系列工具教程
4.Java、後端、架構、阿裡巴巴等大廠最新面試題
生活很美好，明天見～