?目錄

1、引言

2、什么是 RabbitMQ ？

3、RabbitMQ 優(yōu)勢

4、RabbitMQ 整體架構(gòu)剖析

4.1、發(fā)送消息流程

4.2、消費(fèi)消息流程

5、RabbitMQ 應(yīng)用

5.1、廣播

5.2、RPC

VC++常用功能開發(fā)匯總（專欄文章列表，歡迎訂閱，持續(xù)更新...）https://blog.csdn.net/chenlycly/article/details/124272585C++軟件異常排查從入門到精通系列教程（專欄文章列表，歡迎訂閱，持續(xù)更新...）https://blog.csdn.net/chenlycly/article/details/125529931C++軟件分析工具從入門到精通案例集錦（專欄文章正在更新中...）https://blog.csdn.net/chenlycly/article/details/131405795C/C++基礎(chǔ)與進(jìn)階（專欄文章，持續(xù)更新中...）https://blog.csdn.net/chenlycly/category_11931267.html開源組件及數(shù)據(jù)庫技術(shù)（專欄文章，持續(xù)更新中...）https://blog.csdn.net/chenlycly/category_12458859.html網(wǎng)絡(luò)編程與網(wǎng)絡(luò)問題分享（專欄文章，持續(xù)更新中...）https://blog.csdn.net/chenlycly/category_2276111.html

1、引言

? ? ? ?在進(jìn)行系統(tǒng)設(shè)計的時候，各個模塊、服務(wù)器之間為了實(shí)現(xiàn)數(shù)據(jù)的交互，通常是建立連接通過發(fā)送消息來進(jìn)行。如果將他們一一建立連接，就會出現(xiàn)鏈路太多，每一條鏈路都必須感知對端等問題。此場景下消息將非常混亂，后期維護(hù)也將非常痛苦。為了解決這個問題，精簡系統(tǒng)，引入RabbitMq。各相關(guān)模塊不在相互發(fā)送消息，而將消息都發(fā)送給RabbitMQ，由RabbitMQ負(fù)責(zé)將消息傳遞出去。

? ? ? ?那么，什么是RabbitMQ？RabbitMQ又是如何實(shí)現(xiàn)這些功能的呢？ ??

? ? ? ?在這里，給大家重點(diǎn)推薦一下我的兩個熱門暢銷專欄：

專欄1：（該專欄訂閱量接近350個，有很強(qiáng)的實(shí)戰(zhàn)參考價值，廣受好評?。?/span>

C++軟件異常排查從入門到精通系列教程（專欄文章列表，歡迎訂閱，持續(xù)更新...）https://blog.csdn.net/chenlycly/article/details/125529931

本專欄根據(jù)近幾年C++軟件異常排查的項目實(shí)踐，系統(tǒng)地總結(jié)了引發(fā)C++軟件異常的常見原因以及排查C++軟件異常的常用思路與方法，詳細(xì)講述了C++軟件的調(diào)試方法與手段，以圖文并茂的方式給出具體的實(shí)戰(zhàn)問題分析實(shí)例，帶領(lǐng)大家逐步掌握C++軟件調(diào)試與異常排查的相關(guān)技術(shù)，適合基礎(chǔ)進(jìn)階和想做技術(shù)提升的相關(guān)C++開發(fā)人員！

專欄中的文章都是通過項目實(shí)戰(zhàn)總結(jié)出來的（通過項目實(shí)戰(zhàn)積累了大量的異常排查素材和案例），有很強(qiáng)的實(shí)戰(zhàn)參考價值！專欄文章還在持續(xù)更新中，預(yù)計文章篇數(shù)能更新到200篇以上！

專欄2：?

C/C++基礎(chǔ)與進(jìn)階https://blog.csdn.net/chenlycly/category_11931267.html

以多年的開發(fā)實(shí)戰(zhàn)為基礎(chǔ)，總結(jié)并講解一些的C/C++基礎(chǔ)與進(jìn)階內(nèi)容，以圖文并茂的方式對相關(guān)知識點(diǎn)進(jìn)行詳細(xì)地展開與闡述！專欄涉及了C/C++領(lǐng)域的多個方面的內(nèi)容，同時給出C/C++及網(wǎng)絡(luò)方面的常見筆試面試題，并詳細(xì)講述Visual Studio常用調(diào)試手段與技巧！

2、什么是 RabbitMQ ？

? ? ? ?在講RabbitMQ之前，需要先了解一下AMQP的概念。

? ? ? ?AMQP，即Advanced Message Queuing Protocol（高級消息隊列協(xié)議），是一個提供統(tǒng)一消息服務(wù)的應(yīng)用層標(biāo)準(zhǔn)高級消息隊列協(xié)議。AMQP是應(yīng)用層協(xié)議的一個開放標(biāo)準(zhǔn)，為面向消息的中間件設(shè)計，基于此協(xié)議的客戶端與消息中間件傳遞消息，不受客戶端/中間件不同產(chǎn)品、不同開發(fā)語言等條件的限制。該協(xié)議是一種二進(jìn)制協(xié)議，提供客戶端應(yīng)用于消息中間件之間異步、安全、高效的交互。相對于我們常見的REST API，AMQP更容易實(shí)現(xiàn)，可以降低開銷，同時靈活性高，可以輕松的添加負(fù)載平衡和高可用性的功能，并保證消息傳遞，在性能上AMQP協(xié)議也相對更好一些。

? ? ? ?RabbitMQ是AMQP的一個開源實(shí)現(xiàn)，服務(wù)器端用Erlang語言編寫，用于在分布式系統(tǒng)中存儲轉(zhuǎn)發(fā)消息，支持多種客戶端，如：Python、Ruby、.NET、Java、JMS、C、PHP、 ActionScript、XMPP、STOMP等，支持AJAX。 MQ（Messages Queue）是一種應(yīng)用程序與應(yīng)用程的通信方法。RabbitMQ相當(dāng)于生產(chǎn)者與消費(fèi)者的模式，消息發(fā)送端（生產(chǎn)者）將消息寫入消息隊列，消息接收端（消費(fèi)者）從消息隊列中取出消息、消費(fèi)消息；而消息的發(fā)送端無需知道消息接受端的存在，反之亦然。

3、RabbitMQ 優(yōu)勢

? ? ? ? RabbitMQ主要有以下幾個優(yōu)勢：

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

4、RabbitMQ整體架構(gòu)剖析

? ? ? ? 在詳細(xì)介紹RabbitMQ之前，先介紹幾個重要的概念：

• Queue：消息隊列
• Exchange：交換機(jī)，它會按照路由規(guī)則來投遞消息
• Routing key：路由關(guān)鍵字，exchange會根據(jù)它來進(jìn)行消息投遞
• Bind：綁定了queue和exchange，根據(jù)路由規(guī)則將消息會投遞到對應(yīng)的消息隊列中去。
• Producer：消息生產(chǎn)者
• Consumer：消息的消費(fèi)者

? ? ? ?RabbitMQ的整體架構(gòu)圖如下所示：

P（Producer，消息生產(chǎn)者）負(fù)責(zé)發(fā)送，C（Consumer，消息消費(fèi)者）負(fù)責(zé)消費(fèi)消息。其中交換機(jī)exchange、隊列Queue的定義、exchange與Queue的綁定既可以放在發(fā)送端，也可以放在消費(fèi)端，但是不管放在何處定義，要在使用前定義，否則會出錯。本文統(tǒng)一將exchange放在生產(chǎn)者端來定義，而將queue的定義，queue與exchange的綁定放在消費(fèi)端來處理。另外，為了防止第一次使用exchange是在消費(fèi)端，可以在消費(fèi)端也同時定義exchange。本文不考慮這種情況，默認(rèn)在消費(fèi)端使用exchange的時候已定義過。

4.1、發(fā)送消息流程

? ? ? P端發(fā)送消息的基本過程是：

1）連接服務(wù)器；
2）聲明exchange，并設(shè)置其相關(guān)屬性；
3）將消息發(fā)送到exchange。

其中，exchange有3種類型：fanout、routing、topic：

1）fanout不處理路由鍵，為空即可，只要簡單的將隊列綁定到交換機(jī)上，那么發(fā)送到交換機(jī)上的消息都會被轉(zhuǎn)發(fā)到與該交換機(jī)綁定的所有隊列上。
2）Routing處理路由鍵，需要將一個隊列綁定到交換機(jī)上，要求消息與一個特定的路由鍵完全匹配。
3）Topic將路由鍵與某模式進(jìn)行匹配，此時隊列需要綁定到一個模式上。匹配的規(guī)格是”#”匹配一個或多個詞，”*”匹配一個詞。

4.2、消費(fèi)消息流程

? ? ? C消費(fèi)消息的基本過程是：

1）連接服務(wù)器；
2）聲明隊列queue及其屬性（持久化、無消費(fèi)者時是否自動刪除隊列等等）；
3）設(shè)置routingkey，并且通過routingkey將queue與exchange綁定到一起；
4）等待消息，消費(fèi)消息。

其中，queue可以設(shè)置的屬性有：Exclusive、auto_delete、durable。

1）Exclusive：排他隊列，如果一個隊列被聲明為排他隊列，該隊列僅對首次聲明它的連接可見，并在連接斷開時自動刪除。
2）Auto_delete：自動刪除，如果該隊列沒有任何訂閱的消費(fèi)者的話，該隊列會被自動刪除。這種隊列適用于臨時隊列。
3）Durable：服務(wù)器重啟后，隊列不會丟失。?

? ? ? 對上述的exchange、queue、binding的一個例子：

Mq.queue_bind(“QueueTest”, “ExchangeTest”, “Test”)

這個綁定的意思是：任何發(fā)送到交換機(jī)ExchangeTest的具有路由鍵Test的消息都會被路由到名為QueueTest的隊列中。

5、RabbitMQ 應(yīng)用

? ? ? ?一般平臺的消息大致分為兩種類型：notif和req-ack-notif。對應(yīng)于rabbitmq正好有兩種模型：publish/subscribe和rpc。下面根據(jù)實(shí)際應(yīng)用來講解這兩個模型。

5.1、廣播

? ? ? 假設(shè)應(yīng)用服務(wù)器收到了一條消息A，需要廣播給其他多個業(yè)務(wù)服務(wù)器。按照圖一中rabbitmq的基本結(jié)構(gòu)我們應(yīng)該能想到兩種方式：

Method1

Method2

上述兩種方法哪一種能實(shí)現(xiàn)我們的目的？答案是Method1，如果采用Method2的話，queue會將消息依次分發(fā)給兩個消費(fèi)端，例如客戶端C1收到消息1,3,5…，客戶端C2收到消息2,4,6…。

? ? ? ?雖然此種方法不能實(shí)現(xiàn)我們的目的，但在此處插入一點(diǎn)，及每條消息的處理量可能而且?guī)缀蹩隙ㄊ遣煌模杂袝r會出現(xiàn)客戶端C1處理完了N條消息，但客戶端C2一條還沒處理完，為了解決這個問題，rabbitmq提供了公平調(diào)度的概念即Fair dispatch:Rabbitmq不會在同一時間給工作者分配多個任務(wù)，只有在工作者完成任務(wù)之后，才會再次接收到任務(wù)。

? ? ? ?回到剛才討論的地方，我們已經(jīng)確立了使用Method1來完成該功能，現(xiàn)在根據(jù)該方法進(jìn)行一些簡單的編碼驗(yàn)證（注：驗(yàn)證語言為python）。publish/subscribe模型之P客戶端代碼如下：

import pika

#建立連接
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

#聲明交換機(jī)
channel.exchange_declare(exchange='exchangeTest', type='fanout')

#發(fā)送消息
channel.basic_publish(exchange='exchangeTest', routing_key='', body='Hello World!')
connection.close()

publish/subscribe模型之C客戶端代碼：

import pika

#建立連接
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

#創(chuàng)建queue
channel.queue_declare(queue=’QueueTest’)

#綁定
channel.queue_bind(exchange=’exchangeTest’, queue=’QueueTest’)
def callback(ch, method, properties, body):?? ?
print “ [x] Received %r” %(body, )?? ?
channel.basic_consume(callback, queue =’QueueTest’, no_ack=True)
channel.start_consuming()

? ? ? ?AMQP支持在一個TCP連接上啟用多個MQ通信channel，每個channel都可以被應(yīng)用作為通信流, 被分配了一個整數(shù)標(biāo)識，自動由Connection()類的.channel()方法維護(hù)。每個AMQP程序至少要有一個連接和一個channel。

5.2、RPC

? ? ? ?對于大部分消息我們不僅僅是通知，更多的是需要對方在接收到消息后給我們回復(fù)的。此時，
我們就需要rabbitmq提供的RPC模型，如下圖所示：

? ? ? ?RPC模型與廣播模型相比，最大的區(qū)別是消費(fèi)者客戶端在接收到消息的時候，需要給發(fā)送者P回復(fù)消息。而同樣的，消息生產(chǎn)者P也不僅僅是做為發(fā)送端了，他還需要接收來自消費(fèi)端C回復(fù)的消息。

? ? ? ?由P到C我們知道直接將Queue1綁定到exchange上就OK了，那么C回復(fù)消息的時候通過什么回給P呢？為此，rabbitmq在P發(fā)送消息的時候，提供設(shè)置回調(diào)隊列及關(guān)聯(lián)ID，C在給P回復(fù)消息的時候，通過回調(diào)隊列即可。提供關(guān)聯(lián)ID的目的是即使P端收到Queue2的消息，也要驗(yàn)證Correlation_Id是否匹配，不匹配的話，直接忽略。

? ? ? ?使用如下的代碼進(jìn)行驗(yàn)證（注：驗(yàn)證語言為python），RPC模型之P端的代碼如下：

import pika
class Center(object):
? ? def __init__(self):
? ? ? ? self.connection = pika.BlockingConnection(pika.ConnectionParameters(?
host='localhost'))
? ? ? ? self.channel = self.connection.channel() ? ? ?
? ? ? ? #定義接收返回消息的隊列，此處為一隨機(jī)生成的隊列
? ? ? ? result = self.channel.queue_declare(exclusive=True)
? ? ? ? self.callback_queue = result.method.queue
?? ??? ?#等待接收消息
? ? ? ? self.channel.basic_consume(self.on_response, no_ack=True,queue=self.callback_queue)

? ? #定義接收到返回消息的處理方法
? ? def on_response(self, ch, method, props, body):
? ? ? ? self.response = body
? ? def request(self, n):
? ? ? ? self.response = None

#發(fā)送計算請求 ? ? ? ?
self.channel.basic_publish(exchange='',
?routing_key='compute_queue', properties=pika.BasicProperties
(reply_to = self.callback_queue,), body=str(n))
? ? ? ? #接收返回的數(shù)據(jù)
? ? ? ? while self.response is None:
? ? ? ? ? ? self.connection.process_data_events()
? ? ? ? return int(self.response)
center = Center()
response = center.request(30)
print " [.] Got %r" % (response,)

RPC模型之C端代碼：文章來源地址http://www.zghlxwxcb.cn/news/detail-753933.html

import pika

class Center(object):
? ? def __init__(self):
? ? ? ? self.connection = pika.BlockingConnection(pika.ConnectionParameters(?
host='localhost'))
? ? ? ? self.channel = self.connection.channel() ? ? ?
? ? ? ? #定義接收返回消息的隊列，此處為一隨機(jī)生成的隊列
? ? ? ? result = self.channel.queue_declare(exclusive=True)
? ? ? ? self.callback_queue = result.method.queue

?? ??? ?#等待接收消息
? ? ? ? self.channel.basic_consume(self.on_response, no_ack=True,queue=self.callback_queue)
? ? #定義接收到返回消息的處理方法
? ? def on_response(self, ch, method, props, body):
? ? ? ? self.response = body
? ? def request(self, n):
? ? ? ? self.response = None

#發(fā)送計算請求 ? ? ? ?
self.channel.basic_publish(exchange='',
?routing_key='compute_queue', properties=pika.BasicProperties
(reply_to = self.callback_queue,),body=str(n))
? ? ? ? #接收返回的數(shù)據(jù)
? ? ? ? while self.response is None:
? ? ? ? ? ? self.connection.process_data_events()
? ? ? ? return int(self.response)
center = Center()
response = center.request(30)
print " [.] Got %r" % (response,)

到了這里，關(guān)于深入詳解高性能消息隊列中間件 RabbitMQ的文章就介紹完了。如果您還想了解更多內(nèi)容，請在右上角搜索TOY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！