“我們大家都知道把一個微服務架構變成一個異步架構只需要加一個MQ，現在市面上有很多MQ的開源框架。到底選擇哪一個MQ的開源框架才合適呢?”
一、什么是MQ？MQ的原理是什么？
MQ就是消息隊列，是Message Queue的縮寫。消息隊列是一種通信方式。消息的本質就是一種數據結構。因為MQ把項目中的消息集中式的處理和存儲，所以MQ主要有解耦，並發，和削峰的功能。
1，解耦：
MQ的消息生產者和消費者互相不關心對方是否存在，通過MQ這個中間件的存在，使整個系統達到解耦的作用。
如果服務之間用RPC通信，當一個服務跟幾百個服務通信時，如果那個服務的通信接口改變，那么幾百個服務的通信接口都的跟着變動，這是非常頭疼的一件事。
但是采用MQ之后，不管是生產者或者消費者都可以單獨改變自己。他們的改變不會影響到別的服務。從而達到解耦的目的。為什么要解耦呢？說白了就是方便，減少不必要的工作量。
2，並發
MQ有生產者集群和消費者集群，所以客戶端是億級用戶時，他們都是並行的。從而大大提升響應速度。
3，削峰
因為MQ能存儲的消息量很大，所以他可以把大量的消息請求先存下了，然后再並發的方式慢慢處理。
如果采用RPC通信，每一次請求用調用RPC接口，當請求量巨大的時候，因為RPC的請求是很耗資源的，所以巨大的請求一定會壓垮服務器。
削峰的目的是用戶體驗變好，並且使整個系統穩定。能承受大量請求消息。
二、現在市面上有什么MQ，
重點介紹RocketMQ
現在市面上的MQ有很多，主要有RabbitMQ，ActiveMQ，ZeroMQ，RocketMQ，Kafka等等，這些都是開源的MQ產品。以前很多人推薦使用RabbitMQ，他也是非常好用的MQ產品，這里不做過多的介紹。Kafka也是高吞吐量的老大，我們這里也不介紹。
我們重點介紹一下RocketMQ，RocketMQ是阿里巴巴在2012年開源的分布式消息中間件，目前已經捐贈給Apache軟件基金會，並於並於2017年9月25日成為 Apache 的頂級項目。
作為經歷過多次阿里巴巴雙十一這種“超級工程”的洗禮並有穩定出色表現的國產中間件，以其高性能、低延時和高可靠等特性近年來已經也被越來越多的國內企業使用。
功能概覽圖
可以看見RocketMQ支持定時和延時消息，這是RabbitMQ所沒有的能力。
RocketMQ的物理結構
從這里可以看出，RocketMQ涉及到四大集群，producer，Name Server，Consumer，Broker。
Producer集群：
是生產者集群，負責產生消息，向消費者發送由業務應用程序系統生成的消息，RocketMQ提供三種方式發送消息：同步，異步，單向。
一，普通消息
1，同步原理圖
同步消息關鍵代碼
try {SendResult sendResult = producer.send(msg);// 同步發送消息，只要不拋異常就是成功if (sendResult != null) {System.out.println(new Date() + " Send mq message success. Topic is:" + msg.getTopic() + " msgId is: " + sendResult.getMessageId());}catch (Exception e) {System.out.println(new Date() + " Send mq message failed. Topic is:" + msg.getTopic());e.printStackTrace();}}
2，異步原理圖
異步消息關鍵代碼
producer.sendAsync(msg, new SendCallback() {public void onSuccess(final SendResult sendResult) {// 消費發送成功System.out.println("send message success. topic=" + sendResult.getTopic() + ", msgId=" + sendResult.getMessageId());}public void onException(OnExceptionContext context) {System.out.println("send message failed. topic=" + context.getTopic() + ", msgId=" + context.getMessageId());}});
3，單向（Oneway）發送原理圖
單向只發送，不等待返回，所以速度最快，一般在微秒級，但可能丟失
單向（Oneway）發送消息關鍵代碼
producer.sendOneway(msg);
三種發送消息具體代碼請參考文檔：https://help.aliyun.com/document_detail/29547.html?spm=a2c4g.11186623.6.566.7e49793fuueSlB[1]
二，定時消息和延時消息
發送定時消息關鍵代碼
try {// 定時消息，單位毫秒（ms），在指定時間戳（當前時間之后）進行投遞，例如 2016-03-07 16:21:00 投遞。如果被設置成當前時間戳之前的某個時刻，消息將立刻投遞給消費者。long timeStamp = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").parse("2016-03-07 16:21:00").getTime();msg.setStartDeliverTime(timeStamp);// 發送消息，只要不拋異常就是成功SendResult sendResult = producer.send(msg);System.out.println("MessageId:"+sendResult.getMessageId());}catch (Exception e) {// 消息發送失敗，需要進行重試處理，可重新發送這條消息或持久化這條數據進行補償處理System.out.println(new Date() + " Send mq message failed. Topic is:" + msg.getTopic());e.printStackTrace();}
發送延時消息關鍵代碼
try {// 延時消息，單位毫秒（ms），在指定延遲時間（當前時間之后）進行投遞，例如消息在 3 秒后投遞long delayTime = System.currentTimeMillis() + 3000;// 設置消息需要被投遞的時間 msg.setStartDeliverTime(delayTime);SendResult sendResult = producer.send(msg);// 同步發送消息，只要不拋異常就是成功if (sendResult != null) {System.out.println(new Date() + " Send mq message success. Topic is:" + msg.getTopic() + " msgId is: " + sendResult.getMessageId());}} catch (Exception e) {// 消息發送失敗，需要進行重試處理，可重新發送這條消息或持久化這條數據進行補償處理System.out.println(new Date() + " Send mq message failed. Topic is:" + msg.getTopic());e.printStackTrace();}
注意事項
1，定時和延時消息的 msg.setStartDeliverTime 參數需要設置成當前時間戳之后的某個時刻（單位毫秒）。如果被設置成當前時間戳之前的某個時刻，消息將立刻投遞給消費者。
2，定時和延時消息的 msg.setStartDeliverTime 參數可設置40天內的任何時刻（單位毫秒），超過40天消息發送將失敗。
3，StartDeliverTime 是服務端開始向消費端投遞的時間。 如果消費者當前有消息堆積，那么定時和延時消息會排在堆積消息后面，將不能嚴格按照配置的時間進行投遞。
4，由於客戶端和服務端可能存在時間差，消息的實際投遞時間與客戶端設置的投遞時間之間可能存在偏差。
5，設置定時和延時消息的投遞時間后，依然受 3 天的消息保存時長限制。例如，設置定時消息 5 天后才能被消費，如果第 5 天后一直沒被消費，那么這條消息將在第8天被刪除。
6，除 Java 語言支持延時消息外，其他語言都不支持延時消息。
發布消息原理圖
三，事務消息
RocketMQ提供類似X/Open XA的分布式事務功能來確保業務發送方和MQ消息的最終一致性，其本質是通過半消息的方式把分布式事務放在MQ端來處理。
原理圖
其中：
1，發送方向消息隊列 RocketMQ 服務端發送消息。
2，服務端將消息持久化成功之后，向發送方 ACK 確認消息已經發送成功，此時消息為半消息。
3，發送方開始執行本地事務邏輯。
4，發送方根據本地事務執行結果向服務端提交二次確認（Commit 或是 Rollback），服務端收到 Commit 狀態則將半消息標記為可投遞，訂閱方最終將收到該消息；服務端收到 Rollback 狀態則刪除半消息，訂閱方將不會接受該消息。
5，在斷網或者是應用重啟的特殊情況下，上述步驟 4 提交的二次確認最終未到達服務端，經過固定時間后服務端將對該消息發起消息回查。
6，發送方收到消息回查后，需要檢查對應消息的本地事務執行的最終結果。
7，發送方根據檢查得到的本地事務的最終狀態再次提交二次確認，服務端仍按照步驟 4 對半消息進行操作。
RocketMQ的半消息機制的注意事項是
1，根據第六步可以看出他要求發送方提供業務回查接口。
2，不能保證發送方的消息冪等，在ack沒有返回的情況下，可能存在重復消息
3，消費方要做冪等處理。
核心代碼
final BusinessService businessService = new BusinessService(); // 本地業務
TransactionProducer producer = ONSFactory.createTransactionProducer(properties,new LocalTransactionCheckerImpl());producer.start();Message msg = new Message("Topic", "TagA", "Hello MQ transaction===".getBytes());try {SendResult sendResult = producer.send(msg, new LocalTransactionExecuter() {public TransactionStatus execute(Message msg, Object arg) {// 消息 ID（有可能消息體一樣，但消息 ID 不一樣，當前消息 ID 在控制台無法查詢）String msgId = msg.getMsgID();// 消息體內容進行 crc32，也可以使用其它的如 MD5long crc32Id = HashUtil.crc32Code(msg.getBody());// 消息 ID 和 crc32id 主要是用來防止消息重復// 如果業務本身是冪等的，可以忽略，否則需要利用 msgId 或 crc32Id 來做冪等// 如果要求消息絕對不重復，推薦做法是對消息體 body 使用 crc32 或 MD5 來防止重復消息Object businessServiceArgs = new Object();TransactionStatus transactionStatus =TransactionStatus.Unknow;try {boolean isCommit = businessService.execbusinessService(businessServiceArgs);if (isCommit) {// 本地事務成功則提交消息 transactionStatus = TransactionStatus.CommitTransaction;} else {// 本地事務失敗則回滾消息 transactionStatus = TransactionStatus.RollbackTransaction;}} catch (Exception e) {log.error("Message Id:{}", msgId, e);}System.out.println(msg.getMsgID());log.warn("Message Id:{}transactionStatus:{}", msgId, transactionStatus.name());return transactionStatus;}}, null);}catch (Exception e) {// 消息發送失敗，需要進行重試處理，可重新發送這條消息或持久化這條數據進行補償處理System.out.println(new Date() + " Send mq message failed. Topic is:" + msg.getTopic());e.printStackTrace();}
具體代碼參考文檔：https://help.aliyun.com/document_detail/29548.html?spm=a2c4g.11186623.6.570.5d5738a49FJl1t[2]
所有消息發布原理圖
producer完全無狀態，可以集群部署。
Name Server集群：
NameServer是一個幾乎無狀態的節點，可集群部署，節點之間無任何信息同步，NameServer很像注冊中心的功能。
聽說阿里之前的NameServer 是用ZooKeeper做的，可能因為Zookeeper不能滿足大規模並發的要求，所以之后NameServer 是阿里自研的。
NameServer其實就是一個路由表，他管理Producer和Comsumer之間的發現和注冊。
Broker集群：
Broker部署相對復雜，Broker分為Master與Slave，一個Master可以對應多個Slaver，但是一個Slaver只能對應一個Master，Master與Slaver的對應關系通過指定相同的BrokerName。
不同的BrokerId來定義，BrokerId為0表示Master，非0表示Slaver。Master可以部署多個。每個Broker與NameServer集群中的所有節點建立長連接，定時注冊Topic信息到所有的NameServer。
Consumer集群：
訂閱方式
消息隊列 RocketMQ 支持以下兩種訂閱方式：
集群訂閱：同一個 Group ID 所標識的所有 Consumer 平均分攤消費消息。 例如某個 Topic 有 9 條消息，一個 Group ID 有 3 個 Consumer 實例，那么在集群消費模式下每個實例平均分攤，只消費其中的 3 條消息。
// 集群訂閱方式設置（不設置的情況下，默認為集群訂閱方式）properties.put(PropertyKeyConst.MessageModel, PropertyValueConst.CLUSTERING);
廣播訂閱：同一個 Group ID 所標識的所有 Consumer 都會各自消費某條消息一次。 例如某個 Topic 有 9 條消息，一個 Group ID 有 3 個 Consumer 實例，那么在廣播消費模式下每個實例都會各自消費 9 條消息。
// 廣播訂閱方式設置properties.put(PropertyKeyConst.MessageModel, PropertyValueConst.BR
OADCASTING);
訂閱消息關鍵代碼：
Consumer consumer = ONSFactory.createConsumer(properties);consumer.subscribe("TopicTestMQ", "TagA||TagB", **new** MessageListener() { //訂閱多個 Tagpublic Action consume(Message message, ConsumeContext context) {System.out.println("Receive: " + message);return Action.CommitMessage;}});//訂閱另外一個 Topicconsumer.subscribe("TopicTestMQ-Other", "*", **new** MessageListener() { //訂閱全部 Tagpublic Action consume(Message message, ConsumeContext context) {System.out.println("Receive: " + message);return Action.CommitMessage;}});consumer.start();
注意事項：
消費端要做冪等處理，所有MQ基本上都不會做冪等處理，需要業務端處理，原因是如果在MQ端做冪等處理會帶來MQ的復雜度，而且嚴重影響MQ的性能。
消息收發模型
主子賬號創建
創建主子賬號的原因是權限問題。下面是主賬號創建流程圖
詳細操作地址：https://help.aliyun.com/document_detail/34411.html?spm=a2c4g.11186623.6.555.38c57f91JXUK7o[3]
子賬號流程圖
詳細操作地址：https://help.aliyun.com/document_detail/96402.html?spm=a2c4g.11186623.6.556.60194fedfSkxIB[4]
三、MQ是微服務架構
非常重要的部分
MQ的誕生把原來的同步架構思維轉變到異步架構思維提供一種方法，為大規模，高並發的業務場景的穩定性實現提供了很好的解決思路。
Martin Fowler強調：分布式調用的第一原則就是不要分布式。這句話看似頗具哲理，然而就企業應用系統而言，只要整個系統在不停地演化，並有多個子系統共同存在時，這條原則就會被迫打破。
Martin Fowler提出的這條原則，一方面是希望設計者能夠審慎地對待分布式調用，另一方面卻也是分布式系統自身存在的缺陷所致。
所以微服務並不是萬能葯，適合的架構才是最好的架構。
Java 極客技術公眾號，是由一群熱愛 Java 開發的技術人組建成立，專注分享原創、高質量的 Java 文章。如果您覺得我們的文章還不錯，請幫忙贊賞、在看、轉發支持，鼓勵我們分享出更好的文章。
關注公眾號，大家可以在公眾號后台回復“博客園”，免費獲得作者 Java 知識體系/面試必看資料。
