RabbitMQ場景觀察

搜索一下很容易得到如下的結果:

異步處理 應用解耦 流量削峰 為何用消息隊列 從上面的描述中可以看出消息隊列是一種應用間的異步協(xié)作機制，那什么時候需要使用?MQ?呢？ 以常見的訂單系統(tǒng)為例，用戶點擊【下單】按鈕之后的業(yè)務邏輯可能包括：扣減庫存、生成相應單據(jù)、發(fā)紅包、發(fā)短信通知。在業(yè)務發(fā)展初期這些邏輯可能放在一起同步執(zhí)行，隨著業(yè)務的發(fā)展訂單量增長，需要提升系統(tǒng)服務的性能，這時可以將一些不需要立即生效的操作拆分出來異步執(zhí)行，比如發(fā)放紅包、發(fā)短信通知等。這種場景下就可以用?MQ?，在下單的主流程（比如扣減庫存、生成相應單據(jù)）完成之后發(fā)送一條消息到?MQ?讓主流程快速完結，而由另外的單獨線程拉取MQ的消息（或者由?MQ?推送消息），當發(fā)現(xiàn)?MQ?中有發(fā)紅包或發(fā)短信之類的消息時，執(zhí)行相應的業(yè)務邏輯。 以上是用于業(yè)務解耦的情況，其它常見場景包括最終一致性、廣播、錯峰流控等等。

那如何和自己的項目場景產生聯(lián)系呢,上面的描述還是籠統(tǒng)了些,在看看一些別人描述的一些比較具體的場景:

消息中間件系列五：RabbitMQ的使用場景（異步處理、應用解耦）

用戶注冊場景:

串行

saveUser.saveUser(user); sendEmail.sendEmail(user.getEmail()); sendSms.sendSms(user.getPhoneNumber()); return?true;

并行

saveUser.saveUser(user); new?Thread(sendEmailFuture).start();???????????? new?Thread(sendSmsFuture).start(); sendEmailFuture.get();//獲取郵件發(fā)送的結果 sendSmsFuture.get();//獲取短信發(fā)送的結果 return?true;

rapbitrq

saveUser.saveUser(user); rabbitTemplate.send("user-reg-exchange","email",??new?Message(user.getEmail().getBytes(),new?MessageProperties())); rabbitTemplate.send("user-reg-exchange","sms",?new?Message(user.getEmail().getBytes(),new?MessageProperties()));

然后測試結果:

串行模式??************spend time : 251ms(這個慢好理解100+100+50)

并行模式??************spend time : 153ms(這個也好理解100+50)

消息隊列模式：************spend time : 59ms （這個是快，但是具體的email和sms其實是沒有確認的,只是解了耦合加快了消息響應,但是如果需要等待一些返回的時候,那么時間會和并行模式的相似）

訂單場景:

用戶下訂單買商品，訂單成功了，去扣減庫存，庫存必須扣減完成，沒有庫存，庫存低于某個閾值，可以扣減成功，要通知其他系統(tǒng)（如采購系統(tǒng)盡快采購，用戶訂單系統(tǒng)我們盡快調貨）

RPC實現(xiàn)。庫存系統(tǒng)失敗，訂單系統(tǒng)也無法成功，訂單系統(tǒng)和庫存系統(tǒng)耦合了。所以要緩存消息中間件來解耦。發(fā)送一個扣減庫存的消息，保證消息必須被庫存系統(tǒng)處理。

三個問題要解決：

1）訂單系統(tǒng)發(fā)給MQ服務器的消息，必須被MO服務器接收到（事物、發(fā)送者確認）

2）MQ服務器拿到消息以后，消息被正常處理以前必須保存住（持久化）

3）某個庫存服務出現(xiàn)了異常，消息要能夠被其他庫存系統(tǒng)處理（消費者確認，消息監(jiān)聽類要實現(xiàn)ChannelAwareMessageListener）

訂單系統(tǒng)一定要知道庫存系統(tǒng)是否處理成功怎么辦？

庫存系統(tǒng)和訂單系統(tǒng)之間建立一個消息通道，庫存系統(tǒng)去通知訂單系統(tǒng)

演示的代碼挺多的，我摘了一些:

channel.basicAck(message.getMessageProperties().getDeliveryTag(),?false);

channel.basicNack(message.getMessageProperties().getDeliveryTag(),?false,true);

--->回想一下這種場景，如果不是有rabbitRq會怎么處理呢,非分布式的場景下,庫存失敗訂單失敗,通過數(shù)據(jù)庫事務可以達到一定程度的效果,但是庫存系統(tǒng)和訂單系統(tǒng)勢必是非完全隔離的；通過數(shù)據(jù)庫事件或者定時任務不斷檢測庫存來通知其他系統(tǒng)，分布式的場景下也可以采用類似的邏輯，這樣處理即時性不好，而且當檢測周期設置的比較短的時也會占用系統(tǒng)資源影響效率--->額,但是功能上應該沒問題,非身臨其境不好把握問題的關鍵啊

假設預設前提,庫存系統(tǒng)和訂單系統(tǒng)是完全各自獨立運行的包括數(shù)據(jù),通過rpc或者rest或者soap之類的進行一些同步的交流,和上面一樣同步串行導致部分耦合,而rabbitrq是異步的消息隊列

參考:

RabbitMQ 的應用場景以及基本原理介紹

摘一部分:

RabbitMQ 特性

RabbitMQ 最初起源于金融系統(tǒng)，用于在分布式系統(tǒng)中存儲轉發(fā)消息，在易用性、擴展性、高可用性等方面表現(xiàn)不俗。具體特點包括：

-?可靠性（Reliability） RabbitMQ?使用一些機制來保證可靠性，如持久化、傳輸確認、發(fā)布確認。 ?-?靈活的路由（Flexible?Routing） 在消息進入隊列之前，通過?Exchange?來路由消息的。對于典型的路由功能，RabbitMQ?已經提供了一些內置的?Exchange?來實現(xiàn)。針對更復雜的路由功能，可以將多個?Exchange?綁定在一起，也通過插件機制實現(xiàn)自己的?Exchange?。 ?-?消息集群（Clustering） 多個?RabbitMQ?服務器可以組成一個集群，形成一個邏輯?Broker?。 ?-?高可用（Highly?Available?Queues） 隊列可以在集群中的機器上進行鏡像，使得在部分節(jié)點出問題的情況下隊列仍然可用。 ?-?多種協(xié)議（Multi-protocol） RabbitMQ?支持多種消息隊列協(xié)議，比如?STOMP、MQTT?等等。 ?-?多語言客戶端（Many?Clients） RabbitMQ?幾乎支持所有常用語言，比如?Java、.NET、Ruby?等等。 ?-?管理界面（Management?UI） RabbitMQ?提供了一個易用的用戶界面，使得用戶可以監(jiān)控和管理消息?Broker?的許多方面。 ?-?跟蹤機制（Tracing） 如果消息異常，RabbitMQ?提供了消息跟蹤機制，使用者可以找出發(fā)生了什么。 ?-?插件機制（Plugin?System） RabbitMQ?提供了許多插件，來從多方面進行擴展，也可以編寫自己的插件。

RabbitMQ 基本概念

-?Message 消息，消息是不具名的，它由消息頭和消息體組成。消息體是不透明的，而消息頭則由一系列的可選屬性組成，這些屬性包括routing-key（路由鍵）、priority（相對于其他消息的優(yōu)先權）、delivery-mode（指出該消息可能需要持久性存儲）等。 ?-?Publisher 消息的生產者，也是一個向交換器發(fā)布消息的客戶端應用程序。 ?-?Exchange 交換器，用來接收生產者發(fā)送的消息并將這些消息路由給服務器中的隊列。 ?-?Routing?Key 路由關鍵字,exchange根據(jù)這個關鍵字進行消息投遞。 ?-?Binding 綁定，用于消息隊列和交換器之間的關聯(lián)。一個綁定就是基于路由鍵將交換器和消息隊列連接起來的路由規(guī)則，所以可以將交換器理解成一個由綁定構成的路由表。 ?-?Queue 消息隊列，用來保存消息直到發(fā)送給消費者。它是消息的容器，也是消息的終點。一個消息可投入一個或多個隊列。消息一直在隊列里面，等待消費者連接到這個隊列將其取走。 ?-?Connection 網(wǎng)絡連接，比如一個TCP連接。 ?-?Channel 信道，多路復用連接中的一條獨立的雙向數(shù)據(jù)流通道。信道是建立在真實的TCP連接內地虛擬連接，AMQP?命令都是通過信道發(fā)出去的，不管是發(fā)布消息、訂閱隊列還是接收消息，這些動作都是通過信道完成。因為對于操作系統(tǒng)來說建立和銷毀?TCP?都是非常昂貴的開銷，所以引入了信道的概念，以復用一條?TCP?連接。 ?-?Consumer 消息的消費者，表示一個從消息隊列中取得消息的客戶端應用程序。 ?-?Virtual?Host 虛擬主機，表示一批交換器、消息隊列和相關對象。虛擬主機是共享相同的身份認證和加密環(huán)境的獨立服務器域。每個?vhost?本質上就是一個?mini?版的?RabbitMQ?服務器，擁有自己的隊列、交換器、綁定和權限機制。vhost?是?AMQP?概念的基礎，必須在連接時指定，RabbitMQ?默認的?vhost?是?/?。 ?-?Broker

表示消息隊列服務器實體。它提供一種傳輸服務,它的角色就是維護一條從生產者到消費者的路線，保證數(shù)據(jù)能按照指定的方式進行傳輸,

RPC支持

MQ本身是基于異步的消息處理，前面的示例中所有的生產者（P）將消息發(fā)送到RabbitMQ后不會知道消費者（C）處理成功或者失敗（甚至連有沒有消費者來處理這條消息都不知道）。 但實際的應用場景中，我們很可能需要一些同步處理，需要同步等待服務端將我的消息處理完成后再進行下一步處理。這相當于RPC（Remote Procedure Call，遠程過程調用）。在RabbitMQ中也支持RPC。

RabbitMQ 中實現(xiàn)RPC 的機制是：

客戶端發(fā)送請求（消息）時，在消息的屬性（MessageProperties ，在AMQP 協(xié)議中定義了14中properties ，這些屬性會隨著消息一起發(fā)送）中設置兩個值replyTo （一個Queue 名稱，用于告訴服務器處理完成后將通知我的消息發(fā)送到這個Queue 中）和correlationId （此次請求的標識號，服務器處理完成后需要將此屬性返還，客戶端將根據(jù)這個id了解哪條請求被成功執(zhí)行了或執(zhí)行失敗）

服務器端收到消息并處理

服務器端處理完消息后，將生成一條應答消息到replyTo 指定的Queue ，同時帶上correlationId 屬性

客戶端之前已訂閱replyTo 指定的Queue ，從中收到服務器的應答消息后，根據(jù)其中的correlationId 屬性分析哪條請求被執(zhí)行了，根據(jù)執(zhí)行結果進行后續(xù)業(yè)務處理

其他:

https://blog.csdn.net/anzhsoft/article/details/19563091

https://www.cnblogs.com/vipstone/p/9275256.html

https://www.jianshu.com/p/79ca08116d57

https://zhuanlan.zhihu.com/p/48230422+&cd=4&hl=zh-CN&ct=clnk&gl=sg

https://github.com/lemon-china/lemon-rabbitmq

https://www.jianshu.com/p/b5d3a1d6934b

https://blog.csdn.net/zheng_lan_fang/article/details/78612870

https://juejin.im/post/5d39101bf265da1bd3059da6

http://webcache.googleusercontent.com/search?q=cache:CTgcssVhOOMJ:coderec.cn/2016/04/10/%25E4%25BD%25BF%25E7%2594%25A8%25E6%25B6%2588%25E6%2581%25AF%25E9%2598%259F%25E5%2588%2597%25E8%25BF%259B%25E8%25A1%258C%25E8%25A7%25A3%25E8%2580%25A6/+&cd=5&hl=zh-CN&ct=clnk&gl=sg

https://github.com/leeSmall/MessageMiddleware

https://webcache.googleusercontent.com/search?q=cache:xcTN_cT_EegJ:https://zhuanlan.zhihu.com/p/66022936+&cd=1&hl=zh-CN&ct=clnk&gl=sg

版權聲明：本文內容由網(wǎng)絡用戶投稿，版權歸原作者所有，本站不擁有其著作權，亦不承擔相應法律責任。如果您發(fā)現(xiàn)本站中有涉嫌抄襲或描述失實的內容，請聯(lián)系我們jiasou666@gmail.com 處理，核實后本網(wǎng)站將在24小時內刪除侵權內容。