溫馨提示×
您好，登錄后才能下訂單哦！
點擊 登錄注冊 即表示同意《億速云用戶服務條款》
您好，登錄后才能下訂單哦！
本篇內容主要講解“Java消息隊列什么時候使用”，感興趣的朋友不妨來看看。本文介紹的方法操作簡單快捷，實用性強。下面就讓小編來帶大家學習“Java消息隊列什么時候使用”吧!
當你需要使用消息隊列時，首先需要考慮它的必要性。可以使用mq的場景有很多，最常用的幾種，是做業務解耦/最終一致性/廣播/錯峰流控等。反之，如果需要強一致性，關注業務邏輯的處理結果，則RPC顯得更為合適。
解耦是消息隊列要解決的最本質問題。所謂解耦，簡單點講就是一個事務，只關心核心的流程。而需要依賴其他系統但不那么重要的事情，有通知即可，無需等待結果。換句話說，基于消息的模型，關心的是“通知”，而非“處理”。
比如在美團旅游，我們有一個產品中心，產品中心上游對接的是主站、移動后臺、旅游供應鏈等各個數據源；下游對接的是篩選系統、API系統等展示系統。當上游的數據發生變更的時候，如果不使用消息系統，勢必要調用我們的接口來更新數據，就特別依賴產品中心接口的穩定性和處理能力。但其實，作為旅游的產品中心，也許只有對于旅游自建供應鏈，產品中心更新成功才是他們關心的事情。而對于團購等外部系統，產品中心更新成功也好、失敗也罷，并不是他們的職責所在。他們只需要保證在信息變更的時候通知到我們就好了。
而我們的下游，可能有更新索引、刷新緩存等一系列需求。對于產品中心來說，這也不是我們的職責所在。說白了，如果他們定時來拉取數據，也能保證數據的更新，只是實時性沒有那么強。但使用接口方式去更新他們的數據，顯然對于產品中心來說太過于“重量級”了，只需要發布一個產品ID變更的通知，由下游系統來處理，可能更為合理。
再舉一個例子，對于我們的訂單系統，訂單最終支付成功之后可能需要給用戶發送短信積分什么的，但其實這已經不是我們系統的核心流程了。如果外部系統速度偏慢（比如短信網關速度不好），那么主流程的時間會加長很多，用戶肯定不希望點擊支付過好幾分鐘才看到結果。那么我們只需要通知短信系統“我們支付成功了”，不一定非要等待它處理完成。
最終一致性指的是兩個系統的狀態保持一致，要么都成功，要么都失敗。當然有個時間限制，理論上越快越好，但實際上在各種異常的情況下，可能會有一定延遲達到最終一致狀態，但最后兩個系統的狀態是一樣的。
業界有一些為“最終一致性”而生的消息隊列，如Notify（阿里）、QMQ（去哪兒）等，其設計初衷，就是為了交易系統中的高可靠通知。
以一個銀行的轉賬過程來理解最終一致性，轉賬的需求很簡單，如果A系統扣錢成功，則B系統加錢一定成功。反之則一起回滾，像什么都沒發生一樣。
然而，這個過程中存在很多可能的意外：
A扣錢成功，調用B加錢接口失敗。
A扣錢成功，調用B加錢接口雖然成功，但獲取最終結果時網絡異常引起超時。
A扣錢成功，B加錢失敗，A想回滾扣的錢，但A機器down機。
可見，想把這件看似簡單的事真正做成，真的不那么容易。所有跨VM的一致性問題，從技術的角度講通用的解決方案是：
強一致性，分布式事務，但落地太難且成本太高，后文會具體提到。
最終一致性，主要是用“記錄”和“補償”的方式。在做所有的不確定的事情之前，先把事情記錄下來，然后去做不確定的事情，結果可能是：成功、失敗或是不確定，“不確定”（例如超時等）可以等價為失敗。成功就可以把記錄的東西清理掉了，對于失敗和不確定，可以依靠定時任務等方式把所有失敗的事情重新搞一遍，直到成功為止。
回到剛才的例子，系統在A扣錢成功的情況下，把要給B“通知”這件事記錄在庫里（為了保證最高的可靠性可以把通知B系統加錢和扣錢成功這兩件事維護在一個本地事務里），通知成功則刪除這條記錄，通知失敗或不確定則依靠定時任務補償性地通知我們，直到我們把狀態更新成正確的為止。
整個這個模型依然可以基于RPC來做，但可以抽象成一個統一的模型，基于消息隊列來做一個“企業總線”。
具體來說，本地事務維護業務變化和通知消息，一起落地（失敗則一起回滾），然后RPC到達broker，在broker成功落地后，RPC返回成功，本地消息可以刪除。否則本地消息一直靠定時任務輪詢不斷重發，這樣就保證了消息可靠落地broker。
broker往consumer發送消息的過程類似，一直發送消息，直到consumer發送消費成功確認。
我們先不理會重復消息的問題，通過兩次消息落地加補償，下游是一定可以收到消息的。然后依賴狀態機版本號等方式做判重，更新自己的業務，就實現了最終一致性。
最終一致性不是消息隊列的必備特性，但確實可以依靠消息隊列來做最終一致性的事情。另外，所有不保證100%不丟消息的消息隊列，理論上無法實現最終一致性。好吧，應該說理論上的100%，排除系統嚴重故障和bug。
像Kafka一類的設計，在設計層面上就有丟消息的可能（比如定時刷盤，如果掉電就會丟消息）。哪怕只丟千分之一的消息，業務也必須用其他的手段來保證結果正確。
消息隊列的基本功能之一是進行廣播。如果沒有消息隊列，每當一個新的業務方接入，我們都要聯調一次新接口。有了消息隊列，我們只需要關心消息是否送達了隊列，至于誰希望訂閱，是下游的事情，無疑極大地減少了開發和聯調的工作量。
比如本文開始提到的產品中心發布產品變更的消息，以及景點庫很多去重更新的消息，可能“關心”方有很多個，但產品中心和景點庫只需要發布變更消息即可，誰關心誰接入。
試想上下游對于事情的處理能力是不同的。比如，Web前端每秒承受上千萬的請求，并不是什么神奇的事情，只需要加多一點機器，再搭建一些LVS負載均衡設備和Nginx等即可。但數據庫的處理能力卻十分有限，即使使用SSD加分庫分表，單機的處理能力仍然在萬級。由于成本的考慮，我們不能奢求數據庫的機器數量追上前端。
這種問題同樣存在于系統和系統之間，如短信系統可能由于短板效應，速度卡在網關上（每秒幾百次請求），跟前端的并發量不是一個數量級。但用戶晚上個半分鐘左右收到短信，一般是不會有太大問題的。如果沒有消息隊列，兩個系統之間通過協商、滑動窗口等復雜的方案也不是說不能實現。但系統復雜性指數級增長，勢必在上游或者下游做存儲，并且要處理定時、擁塞等一系列問題。而且每當有處理能力有差距的時候，都需要單獨開發一套邏輯來維護這套邏輯。所以，利用中間系統轉儲兩個系統的通信內容，并在下游系統有能力處理這些消息的時候，再處理這些消息，是一套相對較通用的方式。
總而言之，消息隊列不是萬能的。對于需要強事務保證而且延遲敏感的，RPC是優于消息隊列的。
對于一些無關痛癢，或者對于別人非常重要但是對于自己不是那么關心的事情，可以利用消息隊列去做。
支持最終一致性的消息隊列，能夠用來處理延遲不那么敏感的“分布式事務”場景，而且相對于笨重的分布式事務，可能是更優的處理方式。
當上下游系統處理能力存在差距的時候，利用消息隊列做一個通用的“漏斗”。在下游有能力處理的時候，再進行分發。
如果下游有很多系統關心你的系統發出的通知的時候，果斷地使用消息隊列吧。
這篇文章的標題很難起，網上一翻全是各種MQ的性能比較，很容易讓人以為我也是這么“粗俗”的人（o(╯□╰)o）。我這篇文章想要表達的是——它們根本不是一個東西，有毛的性能好比較？
Message Queue（MQ），消息隊列中間件。很多人都說：MQ通過將消息的發送和接收分離來實現應用程序的異步和解偶，這個給人的直覺是——MQ是異步的，用來解耦的，但是這個只是MQ的效果而不是目的。MQ真正的目的是為了通訊，屏蔽底層復雜的通訊協議，定義了一套應用層的、更加簡單的通訊協議。一個分布式系統中兩個模塊之間通訊要么是HTTP，要么是自己開發的TCP，但是這兩種協議其實都是原始的協議。HTTP協議很難實現兩端通訊——模塊A可以調用B，B也可以主動調用A，如果要做到這個兩端都要背上WebServer，而且還不支持長連接（HTTP 2.0的庫根本找不到）。TCP就更加原始了，粘包、心跳、私有的協議，想一想頭皮就發麻。MQ所要做的就是在這些協議之上構建一個簡單的“協議”——生產者/消費者模型。MQ帶給我的“協議”不是具體的通訊協議，而是更高層次通訊模型。它定義了兩個對象——發送數據的叫生產者；消費數據的叫消費者， 提供一個SDK讓我們可以定義自己的生產者和消費者實現消息通訊而無視底層通訊協議。
列出功能表來比較MQ差異或者來一場“MQ性能大比武”的做法都是比較扯的，首先要做的事情應該是分類。我理解的MQ分為兩個流派
這個流派通常有一臺服務器作為Broker，所有的消息都通過它中轉。生產者把消息發送給它就結束自己的任務了，Broker則把消息主動推送給消費者（或者消費者主動輪詢）。
重Topic流
kafka、JMS就屬于這個流派，生產者會發送key和數據到Broker，由Broker比較key之后決定給那個消費者。這種模式是我們最常見的模式，是我們對MQ最多的印象。在這種模式下一個topic往往是一個比較大的概念，甚至一個系統中就可能只有一個topic，topic某種意義上就是queue，生產者發送key相當于說：“hi，把數據放到key的隊列中”。
如上圖所示，Broker定義了三個隊列，key1，key2，key3，生產者發送數據的時候會發送key1和data，Broker在推送數據的時候則推送data（也可能把key帶上）。雖然架構一樣但是kafka的性能要比jms的性能不知道高到多少倍，所以基本這種類型的MQ只有kafka一種備選方案。如果你需要一條暴力的數據流（在乎性能而非靈活性）那么kafka是最好的選擇。
輕Topic流
這種的代表是RabbitMQ（或者說是AMQP）。生產者發送key和數據，消費者定義訂閱的隊列，Broker收到數據之后會通過一定的邏輯計算出key對應的隊列，然后把數據交給隊列。
注意到了嗎？這種模式下解耦了key和queue，在這種架構中queue是非常輕量級的（在RabbitMQ中它的上限取決于你的內存），消費者關心的只是自己的queue；生產者不必關心數據最終給誰只要指定key就行了，中間的那層映射在AMQP中叫exchange（交換機）。AMQP中有四種種exchange——Direct exchange：key就等于queue；Fanout exchange：無視key，給所有的queue都來一份；Topic exchange：key可以用“寬字符”模糊匹配queue；最后一個厲害了Headers exchange：無視key，通過查看消息的頭部元數據來決定發給那個queue（AMQP頭部元數據非常豐富而且可以自定義）。這種結構的架構給通訊帶來了很大的靈活性，我們能想到的通訊方式都可以用這四種exchange表達出來。如果你需要一個企業數據總線（在乎靈活性）那么RabbitMQ絕對的值得一用。
此門派是AMQP的“叛徒”，某位道友嫌棄AMQP太“重”（那是他沒看到用Erlang實現的時候是多么的行云流水） 所以設計了zeromq。這位道友非常睿智，他非常敏銳的意識到——MQ是更高級的Socket，它是解決通訊問題的。所以ZeroMQ被設計成了一個“庫”而不是一個中間件，這種實現也可以達到——沒有broker的目的。
節點之間通訊的消息都是發送到彼此的隊列中，每個節點都既是生產者又是消費者。ZeroMQ做的事情就是封裝出一套類似于scoket的API可以完成發送數據，讀取數據。如果你仔細想一下其實ZeroMQ是這樣的
頓悟了嗎？Actor模型，ZeroMQ其實就是一個跨語言的、重量級的Actor模型郵箱庫。你可以把自己的程序想象成一個actor，zeromq就是提供郵箱功能的庫；zeromq可以實現同一臺機器的IPC通訊也可以實現不同機器的TCP、UDP通訊。如果你需要一個強大的、靈活、野蠻的通訊能力，別猶豫zeromq。
答案是否定了，首先ZeroMQ支持請求->應答模式；其次RabbitMQ提供了RPC是地地道道的同步通訊，只有JMS、kafka這種架構才只能做異步。我們很多人第一次接觸MQ都是JMS之類的這種所以才會產生這種錯覺。
kafka，zeromq，rabbitmq代表了三種完全不同風格的MQ架構；關注點完全不同：
kafka在乎的是性能，速度
rabbitmq追求的是靈活
zeromq追求的是輕量級、分布式
如果你拿zeromq來做大數據量的傳輸功能，不是生產者的內存“爆掉”就是消費者被“壓死”；如果你用kafka做通訊總線那絕對的不會快只能更慢；你想要rabbitmq實現分布式，那真的是難為它。
我們現在明確了消息隊列的使用場景，下一步就是如何設計實現一個消息隊列了。
基于消息的系統模型，不一定需要broker(消息隊列服務端)。市面上的的Akka（actor模型）、ZeroMQ等，其實都是基于消息的系統設計范式，但是沒有broker。
我們之所以要設計一個消息隊列，并且配備broker，無外乎要做兩件事情：
消息的轉儲，在更合適的時間點投遞，或者通過一系列手段輔助消息最終能送達消費機。
規范一種范式和通用的模式，以滿足解耦、最終一致性、錯峰等需求。
掰開了揉碎了看，最簡單的消息隊列可以做成一個消息轉發器，把一次RPC做成兩次RPC。發送者把消息投遞到服務端（以下簡稱broker），服務端再將消息轉發一手到接收端，就是這么簡單。
一般來講，設計消息隊列的整體思路是先build一個整體的數據流,例如producer發送給broker,broker發送給consumer,consumer回復消費確認，broker刪除/備份消息等。
利用RPC將數據流串起來。然后考慮RPC的高可用性，盡量做到無狀態，方便水平擴展。
之后考慮如何承載消息堆積，然后在合適的時機投遞消息，而處理堆積的最佳方式，就是存儲，存儲的選型需要綜合考慮性能/可靠性和開發維護成本等諸多因素。
為了實現廣播功能，我們必須要維護消費關系，可以利用zk/config server等保存消費關系。
在完成了上述幾個功能后，消息隊列基本就實現了。然后我們可以考慮一些高級特性，如可靠投遞，事務特性，性能優化等。
下面我們會以設計消息隊列時重點考慮的模塊為主線，穿插灌輸一些消息隊列的特性實現方法，來具體分析設計實現一個消息隊列時的方方面面。
剛才講到，所謂消息隊列，無外乎兩次RPC加一次轉儲，當然需要消費端最終做消費確認的情況是三次RPC。既然是RPC，就必然牽扯出一系列話題，什么負載均衡啊、服務發現啊、通信協議啊、序列化協議啊，等等。在這一塊，我的強烈建議是不要重復造輪子。利用公司現有的RPC框架：Thrift也好，Dubbo也好，或者是其他自定義的框架也好。因為消息隊列的RPC，和普通的RPC沒有本質區別。當然了，自主利用Memchached或者Redis協議重新寫一套RPC框架并非不可（如MetaQ使用了自己封裝的Gecko NIO框架，卡夫卡也用了類似的協議）。但實現成本和難度無疑倍增。排除對效率的極端要求，都可以使用現成的RPC框架。
簡單來講，服務端提供兩個RPC服務，一個用來接收消息，一個用來確認消息收到。并且做到不管哪個server收到消息和確認消息，結果一致即可。當然這中間可能還涉及跨IDC的服務的問題。這里和RPC的原則是一致的，盡量優先選擇本機房投遞。你可能會問，如果producer和consumer本身就在兩個機房了，怎么辦？首先，broker必須保證感知的到所有consumer的存在。其次，producer盡量選擇就近的機房就好了。
其實所有的高可用，是依賴于RPC和存儲的高可用來做的。先來看RPC的高可用，美團的基于MTThrift的RPC框架，阿里的Dubbo等，其本身就具有服務自動發現，負載均衡等功能。而消息隊列的高可用，只要保證broker接受消息和確認消息的接口是冪等的，并且consumer的幾臺機器處理消息是冪等的，這樣就把消息隊列的可用性，轉交給RPC框架來處理了。
那么怎么保證冪等呢？最簡單的方式莫過于共享存儲。broker多機器共享一個DB或者一個分布式文件/kv系統，則處理消息自然是冪等的。就算有單點故障，其他節點可以立刻頂上。另外failover可以依賴定時任務的補償，這是消息隊列本身天然就可以支持的功能。存儲系統本身的可用性我們不需要操太多心，放心大膽的交給DBA們吧！
對于不共享存儲的隊列，如Kafka使用分區加主備模式，就略微麻煩一些。需要保證每一個分區內的高可用性，也就是每一個分區至少要有一個主備且需要做數據的同步，關于這塊HA的細節，可以參考下篇pull模型消息系統設計。
消息到達服務端如果不經過任何處理就到接收者了，broker就失去了它的意義。為了滿足我們錯峰/流控/最終可達等一系列需求，把消息存儲下來，然后選擇時機投遞就顯得是順理成章的了。
只是這個存儲可以做成很多方式。比如存儲在內存里，存儲在分布式KV里，存儲在磁盤里，存儲在數據庫里等等。但歸結起來，主要有持久化和非持久化兩種。
持久化的形式能更大程度地保證消息的可靠性（如斷電等不可抗外力），并且理論上能承載更大限度的消息堆積（外存的空間遠大于內存）。
但并不是每種消息都需要持久化存儲。很多消息對于投遞性能的要求大于可靠性的要求，且數量極大（如日志）。這時候，消息不落地直接暫存內存，嘗試幾次failover，最終投遞出去也未嘗不可。
市面上的消息隊列普遍兩種形式都支持。當然具體的場景還要具體結合公司的業務來看。
我們來看看如果需要數據落地的情況下各種存儲子系統的選擇。理論上，從速度來看，文件系統>分布式KV（持久化）>分布式文件系統>數據庫，而可靠性卻截然相反。還是要從支持的業務場景出發作出最合理的選擇，如果你們的消息隊列是用來支持支付/交易等對可靠性要求非常高，但對性能和量的要求沒有這么高，而且沒有時間精力專門做文件存儲系統的研究，DB是最好的選擇。
但是DB受制于IOPS，如果要求單broker 5位數以上的QPS性能，基于文件的存儲是比較好的解決方案。整體上可以采用數據文件+索引文件的方式處理，具體這塊的設計比較復雜，可以參考下篇的存儲子系統設計。
分布式KV（如MongoDB，HBase）等，或者持久化的Redis，由于其編程接口較友好，性能也比較可觀，如果在可靠性要求不是那么高的場景，也不失為一個不錯的選擇。
現在我們的消息隊列初步具備了轉儲消息的能力。下面一個重要的事情就是解析發送接收關系，進行正確的消息投遞了。
市面上的消息隊列定義了一堆讓人暈頭轉向的名詞，如JMS 規范中的Topic/Queue，Kafka里面的Topic/Partition/ConsumerGroup，RabbitMQ里面的Exchange等等。拋開現象看本質，無外乎是單播與廣播的區別。所謂單播，就是點到點；而廣播，是一點對多點。當然，對于互聯網的大部分應用來說，組間廣播、組內單播是最常見的情形。
消息需要通知到多個業務集群，而一個業務集群內有很多臺機器，只要一臺機器消費這個消息就可以了。
當然這不是絕對的，很多時候組內的廣播也是有適用場景的，如本地緩存的更新等等。另外，消費關系除了組內組間，可能會有多級樹狀關系。這種情況太過于復雜，一般不列入考慮范圍。所以，一般比較通用的設計是支持組間廣播，不同的組注冊不同的訂閱。組內的不同機器，如果注冊一個相同的ID，則單播；如果注冊不同的ID(如IP地址+端口)，則廣播。
至于廣播關系的維護，一般由于消息隊列本身都是集群，所以都維護在公共存儲上，如config server、zookeeper等。維護廣播關系所要做的事情基本是一致的:
發送關系的維護。
發送關系變更時的通知。
上面都是些消息隊列基本功能的實現，下面來看一些關于消息隊列特性相關的內容，不管可靠投遞/消息丟失與重復以及事務乃至于性能，不是每個消息隊列都會照顧到，所以要依照業務的需求，來仔細衡量各種特性實現的成本，利弊，最終做出最為合理的設計。
這是個激動人心的話題，完全不丟消息，究竟可不可能？答案是，完全可能，前提是消息可能會重復，并且，在異常情況下，要接受消息的延遲。
方案說簡單也簡單，就是每當要發生不可靠的事情（RPC等）之前，先將消息落地，然后發送。當失敗或者不知道成功失敗（比如超時）時，消息狀態是待發送，定時任務不停輪詢所有待發送消息，最終一定可以送達。
具體來說：
producer往broker發送消息之前，需要做一次落地。
請求到server后，server確保數據落地后再告訴客戶端發送成功。
支持廣播的消息隊列需要對每個待發送的endpoint，持久化一個發送狀態，直到所有endpoint狀態都OK才可刪除消息。
對于各種不確定（超時、down機、消息沒有送達、送達后數據沒落地、數據落地了回復沒收到），其實對于發送方來說，都是一件事情，就是消息沒有送達。
重推消息所面臨的問題就是消息重復。重復和丟失就像兩個噩夢，你必須要面對一個。好在消息重復還有處理的機會，消息丟失再想找回就難了。
Anyway，作為一個成熟的消息隊列，應該盡量在各個環節減少重復投遞的可能性，不能因為重復有解決方案就放縱的亂投遞。
最后說一句，不是所有的系統都要求最終一致性或者可靠投遞，比如一個論壇系統、一個招聘系統。一個重復的簡歷或話題被發布，可能比丟失了一個發布顯得更讓用戶無法接受。不斷重復一句話，任何基礎組件要服務于業務場景。
當broker把消息投遞給消費者后，消費者可以立即響應我收到了這個消息。但收到了這個消息只是第一步，我能不能處理這個消息卻不一定。或許因為消費能力的問題，系統的負荷已經不能處理這個消息；或者是剛才狀態機里面提到的消息不是我想要接收的消息，主動要求重發。
把消息的送達和消息的處理分開，這樣才真正的實現了消息隊列的本質-解耦。所以，允許消費者主動進行消費確認是必要的。當然，對于沒有特殊邏輯的消息，默認Auto Ack也是可以的，但一定要允許消費方主動ack。
對于正確消費ack的，沒什么特殊的。但是對于reject和error，需要特別說明。reject這件事情，往往業務方是無法感知到的，系統的流量和健康狀況的評估，以及處理能力的評估是一件非常復雜的事情。舉個極端的例子，收到一個消息開始build索引，可能這個消息要處理半個小時，但消息量卻是非常的小。所以reject這塊建議做成滑動窗口/線程池類似的模型來控制，
消費能力不匹配的時候，直接拒絕，過一段時間重發，減少業務的負擔。
但業務出錯這件事情是只有業務方自己知道的，就像上文提到的狀態機等等。這時應該允許業務方主動ack error，并可以與broker約定下次投遞的時間。
上文談到重復消息是不可能100%避免的，除非可以允許丟失，那么，順序消息能否100%滿足呢? 答案是可以，但條件更為苛刻：
允許消息丟失。
從發送方到服務方到接受者都是單點單線程。
所以絕對的順序消息基本上是不能實現的，當然在METAQ/Kafka等pull模型的消息隊列中，單線程生產/消費，排除消息丟失，也是一種順序消息的解決方案。
一般來講，一個主流消息隊列的設計范式里，應該是不丟消息的前提下，盡量減少重復消息，不保證消息的投遞順序。
談到重復消息，主要是兩個話題：
如何鑒別消息重復，并冪等的處理重復消息。
一個消息隊列如何盡量減少重復消息的投遞。
先來看看第一個話題，每一個消息應該有它的唯一身份。不管是業務方自定義的，還是根據IP/PID/時間戳生成的MessageId，如果有地方記錄這個MessageId，消息到來是能夠進行比對就
能完成重復的鑒定。數據庫的唯一鍵/bloom filter/分布式KV中的key，都是不錯的選擇。由于消息不能被永久存儲，所以理論上都存在消息從持久化存儲移除的瞬間上游還在投遞的可能（上游因種種原因投遞失敗，不停重試，都到了下游清理消息的時間）。這種事情都是異常情況下才會發生的，畢竟是小眾情況。兩分鐘消息都還沒送達，多送一次又能怎樣呢？冪等的處理消息是一門藝術，因為種種原因重復消息或者錯亂的消息還是來到了，說兩種通用的解決方案：
版本號。
狀態機。
持久性是事務的一個特性，然而只滿足持久性卻不一定能滿足事務的特性。還是拿扣錢/加錢的例子講。滿足事務的一致性特征，則必須要么都不進行，要么都能成功。
解決方案從大方向上有兩種：
兩階段提交，分布式事務。
本地事務，本地落地，補償發送。
分布式事務存在的最大問題是成本太高，兩階段提交協議，對于仲裁down機或者單點故障，幾乎是一個無解的黑洞。對于交易密集型或者I/O密集型的應用，沒有辦法承受這么高的網絡延遲，系統復雜性。
并且成熟的分布式事務一定構建與比較靠譜的商用DB和商用中間件上，成本也太高。
那如何使用本地事務解決分布式事務的問題呢？以本地和業務在一個數據庫實例中建表為例子，與扣錢的業務操作同一個事務里，將消息插入本地數據庫。如果消息入庫失敗，則業務回滾；如果消息入庫成功，事務提交。
然后發送消息（注意這里可以實時發送，不需要等定時任務檢出，以提高消息實時性）。以后的問題就是前文的最終一致性問題所提到的了，只要消息沒有發送成功，就一直靠定時任務重試。
這里有一個關鍵的點，本地事務做的，是業務落地和消息落地的事務，而不是業務落地和RPC成功的事務。這里很多人容易混淆，如果是后者，無疑是事務嵌套RPC，是大忌，會有長事務死鎖等各種風險。
而消息只要成功落地，很大程度上就沒有丟失的風險（磁盤物理損壞除外）。而消息只要投遞到服務端確認后本地才做刪除，就完成了producer->broker的可靠投遞，并且當消息存儲異常時，業務也是可以回滾的。
本地事務存在兩個最大的使用障礙：
配置較為復雜，“綁架”業務方，必須本地數據庫實例提供一個庫表。
對于消息延遲高敏感的業務不適用。
話說回來，不是每個業務都需要強事務的。扣錢和加錢需要事務保證，但下單和生成短信卻不需要事務，不能因為要求發短信的消息存儲投遞失敗而要求下單業務回滾。所以，一個完整的消息隊列應該定義清楚自己可以投遞的消息類型，如事務型消息，本地非持久型消息，以及服務端不落地的非可靠消息等。對不同的業務場景做不同的選擇。另外事務的使用應該盡量低成本、透明化，可以依托于現有的成熟框架，如Spring的聲明式事務做擴展。業務方只需要使用 @Transactional標簽即可。
首先澄清一個概念，異步，同步和oneway是三件事。異步，歸根結底你還是需要關心結果的，但可能不是當時的時間點關心，可以用輪詢或者回調等方式處理結果；同步是需要當時關心
的結果的；而oneway是發出去就不管死活的方式，這種對于某些完全對可靠性沒有要求的場景還是適用的，但不是我們重點討論的范疇。
回歸來看，任何的RPC都是存在客戶端異步與服務端異步的，而且是可以任意組合的：客戶端同步對服務端異步，客戶端異步對服務端異步，客戶端同步對服務端同步，客戶端異步對服務端同步。
對于客戶端來說，同步與異步主要是拿到一個Result，還是Future(Listenable)的區別。實現方式可以是線程池，NIO或者其他事件機制，這里先不展開講。
服務端異步可能稍微難理解一點，這個是需要RPC協議支持的。參考servlet 3.0規范，服務端可以吐一個future給客戶端，并且在future done的時候通知客戶端。
整個過程可以參考下面的代碼：
客戶端同步服務端異步。
Future<Result> future = request(server);//server立刻返回futuresynchronized(future){while(!future.isDone()){ future.wait();//server處理結束后會notify這個future，并修改isdone標志}}return future.get();客戶端同步服務端同步。
Result result = request(server);
客戶端異步服務端同步(這里用線程池的方式)。
Future<Result> future = executor.submit(new Callable(){public void call<Result>(){ result = request(server);}})return future;客戶端異步服務端異步。
Future<Result> future = request(server);//server立刻返回future return future
上面說了這么多，其實是想讓大家脫離兩個誤區：
RPC只有客戶端能做異步，服務端不能。
異步只能通過線程池。
那么，服務端使用異步最大的好處是什么呢？說到底，是解放了線程和I/O。試想服務端有一堆I/O等待處理，如果每個請求都需要同步響應，每條消息都需要結果立刻返回，那么就幾乎沒法做I/O合并
（當然接口可以設計成batch的，但可能batch發過來的仍然數量較少）。而如果用異步的方式返回給客戶端future，就可以有機會進行I/O的合并，把幾個批次發過來的消息一起落地（這種合并對于MySQL等允許batch insert的數據庫效果尤其明顯），并且徹底釋放了線程。不至于說來多少請求開多少線程，能夠支持的并發量直線提高。
來看第二個誤區，返回future的方式不一定只有線程池。換句話說，可以在線程池里面進行同步操作，也可以進行異步操作，也可以不使用線程池使用異步操作（NIO、事件）。
回到消息隊列的議題上，我們當然不希望消息的發送阻塞主流程（前面提到了，server端如果使用異步模型，則可能因消息合并帶來一定程度上的消息延遲），所以可以先使用線程池提交一個發送請求，主流程繼續往下走。
但是線程池中的請求關心結果嗎？Of course，必須等待服務端消息成功落地，才算是消息發送成功。所以這里的模型，準確地說事客戶端半同步半異步（使用線程池不阻塞主流程，但線程池中的任務需要等待server端的返回），server端是純異步。客戶端的線程池wait在server端吐回的future上，直到server端處理完畢，才解除阻塞繼續進行。
總結一句，同步能夠保證結果，異步能夠保證效率，要合理的結合才能做到最好的效率。
上文提到的消息隊列，大多是針對push模型的設計。現在市面上有很多經典的也比較成熟的pull模型的消息隊列，如Kafka、MetaQ等。這跟JMS中傳統的push方式有很大的區別，可謂另辟蹊徑。
我們簡要分析下push和pull模型各自存在的利弊。
慢消費無疑是push模型最大的致命傷，穿成流水線來看，如果消費者的速度比發送者的速度慢很多，勢必造成消息在broker的堆積。假設這些消息都是有用的無法丟棄的，消息就要一直在broker端保存。當然這還不是最致命的，最致命的是broker給consumer推送一堆consumer無法處理的消息，consumer不是reject就是error，然后來回踢皮球。
反觀pull模式，consumer可以按需消費，不用擔心自己處理不了的消息來騷擾自己，而broker堆積消息也會相對簡單，無需記錄每一個要發送消息的狀態，只需要維護所有消息的隊列和偏移量就可以了。所以對于建立索引等慢消費，消息量有限且到來的速度不均勻的情況，pull模式比較合適。
這是pull模式最大的短板。由于主動權在消費方，消費方無法準確地決定何時去拉取最新的消息。如果一次pull取到消息了還可以繼續去pull，如果沒有pull取到則需要等待一段時間重新pull。
但等待多久就很難判定了。你可能會說，我可以有xx動態pull取時間調整算法，但問題的本質在于，有沒有消息到來這件事情決定權不在消費方。也許1分鐘內連續來了1000條消息，然后半個小時沒有新消息產生，
可能你的算法算出下次最有可能到來的時間點是31分鐘之后，或者60分鐘之后，結果下條消息10分鐘后到了，是不是很讓人沮喪？
當然也不是說延遲就沒有解決方案了，業界較成熟的做法是從短時間開始（不會對broker有太大負擔），然后指數級增長等待。比如開始等5ms，然后10ms，然后20ms，然后40ms……直到有消息到來，然后再回到5ms。
即使這樣，依然存在延遲問題：假設40ms到80ms之間的50ms消息到來，消息就延遲了30ms，而且對于半個小時來一次的消息，這些開銷就是白白浪費的。
在阿里的RocketMq里，有一種優化的做法-長輪詢，來平衡推拉模型各自的缺點。基本思路是:消費者如果嘗試拉取失敗，不是直接return,而是把連接掛在那里wait,服務端如果有新的消息到來，把連接notify起來，這也是不錯的思路。但海量的長連接block對系統的開銷還是不容小覷的，還是要合理的評估時間間隔，給wait加一個時間上限比較好~
到此，相信大家對“Java消息隊列什么時候使用”有了更深的了解，不妨來實際操作一番吧！這里是億速云網站，更多相關內容可以進入相關頻道進行查詢，關注我們，繼續學習！
免責聲明：本站發布的內容（圖片、視頻和文字）以原創、轉載和分享為主，文章觀點不代表本網站立場，如果涉及侵權請聯系站長郵箱：is@yisu.com進行舉報，并提供相關證據，一經查實，將立刻刪除涉嫌侵權內容。
