RabbitMQ基礎概念詳細介紹

你是否遇到過兩個（多個）系統間需要通過定時任務來同步某些資料？你是否在為異構系統的不同程序間互相調用、通訊的問題而苦惱、掙紮？如果是，那麼恭喜你，消息服務讓你可以很輕松地解決這些問題。

消息服務擅長于解決多系統、異構系統間的資料交換（消息通知/通訊）問題，你也可以把它用于系統間服務的互相調用（RPC）。本文将要介紹的RabbitMQ就是目前最主流的消息中間件之一。

AMQP，即Advanced Message Queuing Protocol，進階消息隊列協定，是應用層協定的一個開放标準，為面向消息的中間件設計。消息中間件主要用于元件之間的解耦，消息的發送者無需知道消息使用者的存在，反之亦然。

AMQP的主要特征是面向消息、隊列、路由（包括點對點和釋出/訂閱）、可靠性、安全。

RabbitMQ是一個開源的AMQP實作，伺服器端用Erlang語言編寫，支援多種用戶端，如：Python、Ruby、.NET、Java、JMS、C、PHP、ActionScript、XMPP、STOMP等，支援AJAX。用于在分布式系統中存儲轉發消息，在易用性、擴充性、高可用性等方面表現不俗。

下面将重點介紹RabbitMQ中的一些基礎概念，了解了這些概念，是使用好RabbitMQ的基礎。

ConnectionFactory、Connection、Channel都是RabbitMQ對外提供的API中最基本的對象。Connection是RabbitMQ的socket連結，它封裝了socket協定相關部分邏輯。ConnectionFactory為Connection的制造工廠。

Channel是我們與RabbitMQ打交道的最重要的一個接口，我們大部分的業務操作是在Channel這個接口中完成的，包括定義Queue、定義Exchange、綁定Queue與Exchange、釋出消息等。

Queue（隊列）是RabbitMQ的内部對象，用于存儲消息，用下圖表示。

RabbitMQ中的消息都隻能存儲在Queue中，生産者（下圖中的P）生産消息并最終投遞到Queue中，消費者（下圖中的C）可以從Queue中擷取消息并消費。

多個消費者可以訂閱同一個Queue，這時Queue中的消息會被平均分攤給多個消費者進行處理，而不是每個消費者都收到所有的消息并處理。

在實際應用中，可能會發生消費者收到Queue中的消息，但沒有處理完成就當機（或出現其他意外）的情況，這種情況下就可能會導緻消息丢失。為了避免這種情況發生，我們可以要求消費者在消費完消息後發送一個回執給RabbitMQ，RabbitMQ收到消息回執（Message acknowledgment）後才将該消息從Queue中移除；如果RabbitMQ沒有收到回執并檢測到消費者的RabbitMQ連接配接斷開，則RabbitMQ會将該消息發送給其他消費者（如果存在多個消費者）進行處理。這裡不存在timeout概念，一個消費者處理消息時間再長也不會導緻該消息被發送給其他消費者，除非它的RabbitMQ連接配接斷開。

這裡會産生另外一個問題，如果我們的開發人員在處理完業務邏輯後，忘記發送回執給RabbitMQ，這将會導緻嚴重的bug——Queue中堆積的消息會越來越多；消費者重新開機後會重複消費這些消息并重複執行業務邏輯…

另外pub message是沒有ack的。

如果我們希望即使在RabbitMQ服務重新開機的情況下，也不會丢失消息，我們可以将Queue與Message都設定為可持久化的（durable），這樣可以保證絕大部分情況下我們的RabbitMQ消息不會丢失。但依然解決不了小機率丢失事件的發生（比如RabbitMQ伺服器已經接收到生産者的消息，但還沒來得及持久化該消息時RabbitMQ伺服器就斷電了），如果我們需要對這種小機率事件也要管理起來，那麼我們要用到事務。由于這裡僅為RabbitMQ的簡單介紹，是以這裡将不講解RabbitMQ相關的事務。

前面我們講到如果有多個消費者同時訂閱同一個Queue中的消息，Queue中的消息會被平攤給多個消費者。這時如果每個消息的處理時間不同，就有可能會導緻某些消費者一直在忙，而另外一些消費者很快就處理完手頭工作并一直空閑的情況。我們可以通過設定prefetchCount來限制Queue每次發送給每個消費者的消息數，比如我們設定prefetchCount=1，則Queue每次給每個消費者發送一條消息；消費者處理完這條消息後Queue會再給該消費者發送一條消息。

在上一節我們看到生産者将消息投遞到Queue中，實際上這在RabbitMQ中這種事情永遠都不會發生。實際的情況是，生産者将消息發送到Exchange（交換器，下圖中的X），由Exchange将消息路由到一個或多個Queue中（或者丢棄）。

Exchange是按照什麼邏輯将消息路由到Queue的？這個将在Binding一節介紹。

RabbitMQ中的Exchange有四種類型，不同的類型有着不同的路由政策，這将在Exchange Types一節介紹。

生産者在将消息發送給Exchange的時候，一般會指定一個routing key，來指定這個消息的路由規則，而這個routing key需要與Exchange Type及binding key聯合使用才能最終生效。

在Exchange Type與binding key固定的情況下（在正常使用時一般這些内容都是固定配置好的），我們的生産者就可以在發送消息給Exchange時，通過指定routing key來決定消息流向哪裡。

RabbitMQ為routing key設定的長度限制為255 bytes。

RabbitMQ中通過Binding将Exchange與Queue關聯起來，這樣RabbitMQ就知道如何正确地将消息路由到指定的Queue了。

在綁定（Binding）Exchange與Queue的同時，一般會指定一個binding key；消費者将消息發送給Exchange時，一般會指定一個routing key；當binding key與routing key相比對時，消息将會被路由到對應的Queue中。這個将在Exchange Types章節會列舉實際的例子加以說明。

在綁定多個Queue到同一個Exchange的時候，這些Binding允許使用相同的binding key。

binding key 并不是在所有情況下都生效，它依賴于Exchange Type，比如fanout類型的Exchange就會無視binding key，而是将消息路由到所有綁定到該Exchange的Queue。

RabbitMQ常用的Exchange Type有fanout、direct、topic、headers這四種（AMQP規範裡還提到兩種Exchange Type，分别為system與自定義，這裡不予以描述），下面分别進行介紹。

fanout類型的Exchange路由規則非常簡單，它會把所有發送到該Exchange的消息路由到所有與它綁定的Queue中。

上圖中，生産者（P）發送到Exchange（X）的所有消息都會路由到圖中的兩個Queue，并最終被兩個消費者（C1與C2）消費。

direct類型的Exchange路由規則也很簡單，它會把消息路由到那些binding key與routing key完全比對的Queue中。

以上圖的配置為例，我們以routingKey=”error”發送消息到Exchange，則消息會路由到Queue1（amqp.gen-S9b…，這是由RabbitMQ自動生成的Queue名稱）和Queue2（amqp.gen-Agl…）；如果我們以routingKey=”info”或routingKey=”warning”來發送消息，則消息隻會路由到Queue2。如果我們以其他routingKey發送消息，則消息不會路由到這兩個Queue中。

前面講到direct類型的Exchange路由規則是完全比對binding key與routing key，但這種嚴格的比對方式在很多情況下不能滿足實際業務需求。topic類型的Exchange在比對規則上進行了擴充，它與direct類型的Exchage相似，也是将消息路由到binding key與routing key相比對的Queue中，但這裡的比對規則有些不同，它約定：

routing key為一個句點号“. ”分隔的字元串（我們将被句點号“. ”分隔開的每一段獨立的字元串稱為一個單詞），如“stock.usd.nyse”、“nyse.vmw”、“quick.orange.rabbit”

binding key與routing key一樣也是句點号“. ”分隔的字元串

binding key中可以存在兩種特殊字元“*”與“#”，用于做模糊比對，其中“*”用于比對一個單詞，“#”用于比對多個單詞（可以是零個）

以上圖中的配置為例，routingKey=”quick.orange.rabbit”的消息會同時路由到Q1與Q2，routingKey=”lazy.orange.fox”的消息會路由到Q1，routingKey=”lazy.brown.fox”的消息會路由到Q2，routingKey=”lazy.pink.rabbit”的消息會路由到Q2（隻會投遞給Q2一次，雖然這個routingKey與Q2的兩個bindingKey都比對）；routingKey=”quick.brown.fox”、routingKey=”orange”、routingKey=”quick.orange.male.rabbit”的消息将會被丢棄，因為它們沒有比對任何bindingKey。

headers類型的Exchange不依賴于routing key與binding key的比對規則來路由消息，而是根據發送的消息内容中的headers屬性進行比對。

在綁定Queue與Exchange時指定一組鍵值對；當消息發送到Exchange時，RabbitMQ會取到該消息的headers（也是一個鍵值對的形式），對比其中的鍵值對是否完全比對Queue與Exchange綁定時指定的鍵值對；如果完全比對則消息會路由到該Queue，否則不會路由到該Queue。

該類型的Exchange沒有用到過（不過也應該很有用武之地），是以不做介紹。

MQ本身是基于異步的消息處理，前面的示例中所有的生産者（P）将消息發送到RabbitMQ後不會知道消費者（C）處理成功或者失敗（甚至連有沒有消費者來處理這條消息都不知道）。

但實際的應用場景中，我們很可能需要一些同步處理，需要同步等待服務端将我的消息處理完成後再進行下一步處理。這相當于RPC（Remote Procedure Call，遠端過程調用）。在RabbitMQ中也支援RPC。

RabbitMQ中實作RPC的機制是：

用戶端發送請求（消息）時，在消息的屬性（MessageProperties，在AMQP協定中定義了14中properties，這些屬性會随着消息一起發送）中設定兩個值replyTo（一個Queue名稱，用于告訴伺服器處理完成後将通知我的消息發送到這個Queue中）和correlationId（此次請求的辨別号，伺服器處理完成後需要将此屬性返還，用戶端将根據這個id了解哪條請求被成功執行了或執行失敗）

伺服器端收到消息并處理

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

用戶端之前已訂閱replyTo指定的Queue，從中收到伺服器的應答消息後，根據其中的correlationId屬性分析哪條請求被執行了，根據執行結果進行後續業務處理

本文介紹了RabbitMQ中個人認為最重要的概念，充分利用RabbitMQ提供的這些功能就可以處理我們絕大部分的異步業務了。

本文轉自茄子_2008部落格園部落格，原文連結：http://www.cnblogs.com/xd502djj/p/5599074.html，如需轉載請自行聯系原作者。

RabbitMQ基礎概念詳細介紹

繼續閱讀

使用 ctypes 進行 Python 和 C 的混合程式設計

【python】【資料處理】畫多元資料分布圖

NOSQL安全攻擊

mybatis_入門程式Mybatis入門

AOP程式設計_Android優雅權限架構(1)概念基礎，2021金三銀四前言正文大綱正文

php 去掉字元串的最後一個字元及截取原字元串1,2,3,4,5,6,

Effective Java 8:通用程式設計

php——水印

OOM三種類型

工廠模式-三種類型

【python】netconf協定對接管理裝置

「Python 網絡自動化」NETCONF —— Python 使用 NETCONF 管理配置 H3C 網絡裝置

【遞歸】高效率求2的n次幂

win10本地scala和spark安裝安裝scala安裝spark

scala (3) Function 和 Method

在python中建立excel并寫入