.NET 環境中使用RabbitMQ

時間 2019-11-19

原文原文鏈接

在企業應用系統領域，會面對不一樣系統之間的通訊、集成與整合，尤爲當面臨異構系統時，這種分佈式的調用與通訊變得愈加重要。其次，系統中通常會有不少對實時性要求不高的可是執行起來比較較耗時的地方，好比發送短信，郵件提醒，更新文章閱讀計數，記錄用戶操做日誌等等，若是實時處理的話，在用戶訪問量比較大的狀況下，對系統壓力比較大。html

面對這些問題，咱們通常會將這些請求，放在消息隊列中處理；異構系統之間使用消息進行通信。消息傳遞相較文件傳遞與遠程過程調用（RPC）而言，彷佛更勝一籌，由於它具備更好的平臺無關性，並可以很好地支持併發與異步調用。因此若是系統中出現了以下狀況:node

對操做的實時性要求不高，而須要執行的任務極爲耗時；
存在異構系統間的整合；

通常的能夠考慮引入消息隊列。對於第一種狀況，經常會選擇消息隊列來處理執行時間較長的任務。引入的消息隊列就成了消息處理的緩衝區。消息隊列引入的異步通訊機制，使得發送方和接收方都不用等待對方返回成功消息，就能夠繼續執行下面的代碼，從而提升了數據處理的能力。尤爲是當訪問量和數據流量較大的狀況下，就能夠結合消息隊列與後臺任務，經過避開高峯期對大數據進行處理，就能夠有效下降數據庫處理數據的負荷。數據庫

在前面的一篇講解CQRS模式的文章中，全部的對系統的狀態的更改都是經過事件來完成，通常的將事件存儲到消息隊列中，而後進行統一的處理。數組

本文簡單介紹在RabbitMQ這一消息代理工具，以及在.NET中如何使用RabbitMQ.瀏覽器

一環境搭建

首先，因爲RabbitMQ使用Erlang編寫的，須要運行在Erlang運行時環境上，因此在安裝RabbitMQ Server以前須要安裝Erlang 運行時環境，能夠到Erlang官網下載對應平臺的安裝文件。若是沒有安裝運行時環境，安裝RabbitMQ Server的時候，會提示須要先安裝Erlang環境。安裝完成以後，確保已經將Erlang的安裝路徑註冊到系統的環境變量中。安裝完Erlang以後，這個環境會自動設置，若是沒有，在administrator環境下在控制檯下面輸入，也能夠設置：緩存

Setx  ERLANG_HOME 「D:\Program Files (x86)\erl6.3″

而後，去RabbitMQ官網下載RabbitMQ Server服務端程序，選擇合適的平臺版本下載。安裝完成以後，就能夠開始使用了。安全

如今就能夠對RabbitMQ Server進行配置了。服務器

首先，切換到RabbitMQ Server的安裝目錄：cookie

在sbin下面有不少batch文件，用來控制RabbitMQ Server，固然您也能夠直接在安裝開始菜單中來執行相應的操做：網絡

最簡單的方式是使RabbitMQ以Windows Service的方式在後臺運行，因此咱們須要以管理員權限打開cmd，而後切換到sbin目錄下，執行這三條命令便可：

rabbitmq-service install
rabbitmq-service enable
rabbitmq-service start

如今RabbitMQ的服務端已經啓動起來了，rabbitmq 默認端口是 5672。

下面可使用sbin目錄下面的rabbitmqctl.bat這個腳原本查看和控制服務端狀態的，在cmd中直接運行rabbitmqctl status。若是看到如下結果：

顯示node沒有鏈接上，須要到C:\Windows目錄下，將.erlang.cookie文件，拷貝到用戶目錄下 C:\Users\{用戶名}，這是Erlang的Cookie文件，容許與Erlang進行交互，如今重複運行剛纔的命令就會獲得以下信息：

RabbitMQ Server上面也有用戶概念，安裝好以後，使用rabbitmqctl list_users命令，能夠看到上面目前的用戶：

能夠看到，如今只有一個角色爲administrator的名爲guest的用戶，這個是RabbitMQ默認爲咱們建立的，他有RabbitMQ的全部權限，通常的，咱們須要新建一個咱們本身的用戶，設置密碼，並授予權限，並將其設置爲管理員，可使用下面的命令來執行這一操做：

rabbitmqctl  add_user  yy  hello!
rabbitmqctl  set_permissions  yy  ".*"  ".*"  ".*"
rabbitmqctl  set_user_tags yy administrator

上面的一條命令添加了一個名爲yy的用戶，並設置了密碼hello！，下面的命令爲用戶yy分別授予對全部消息隊列的配置、讀和寫的權限。

如今咱們能夠將默認的guest用戶刪掉，使用下面的命令便可：

rabbitmqctl delete_user guest

若是要修改密碼，可使用下面的命令：

rabbitmqctl change_password {username}  {newpassowrd}

二開始使用

在.NET中使用RabbitMQ須要下載RabbitMQ的客戶端程序集，能夠到官網下載，下載解壓後就能夠獲得RabbitMQ.Client.dll，這就是RabbitMQ的客戶端。

在使用RabitMQ以前，須要對下面的幾個基本概念說明一下：

RabbitMQ是一個消息代理。他從消息生產者(producers)那裏接收消息，而後把消息送給消息消費者（consumer）在發送和接受之間，他可以根據設置的規則進行路由，緩存和持久化。

通常提到RabbitMQ和消息，都用到一些專有名詞。

生產(Producing)意思就是發送。發送消息的程序就是一個生產者(producer)。咱們通常用"P"來表示：

隊列(queue)就是郵箱的名稱。消息經過你的應用程序和RabbitMQ進行傳輸，它們只能存儲在隊列（queue）中。隊列（queue）容量沒有限制，你要存儲多少消息均可以——基本上是一個無限的緩衝區。多個生產者（producers）可以把消息發送給同一個隊列，一樣，多個消費者（consumers）也能從同一個隊列（queue）中獲取數據。隊列能夠畫成這樣（圖上是隊列的名稱）：

消費（Consuming）和獲取消息是同樣的意思。一個消費者（consumer）就是一個等待獲取消息的程序。咱們把它畫做"C"：

一般，消息生產者，消息消費者和消息代理不在同一臺機器上。

2.1 Hello World

爲了展現RabbitMQ的基本使用，咱們發送一個HelloWorld消息，而後接收並處理。

首先建立一個控制檯程序，用來將消息發送到RabbitMQ的消息隊列中，代碼以下：

    static void Main(string[] args) { var factory = new ConnectionFactory(); factory.HostName = "localhost"; factory.UserName = "yy"; factory.Password = "hello!"; using (var connection = factory.CreateConnection()) { using (var channel = connection.CreateModel()) { channel.QueueDeclare("hello", false, false, false, null); string message = "Hello World"; var body = Encoding.UTF8.GetBytes(message); channel.BasicPublish("", "hello", null, body); Console.WriteLine(" set {0}", message); } } }

首先，須要建立一個ConnectionFactory，設置目標，因爲是在本機，因此設置爲localhost，若是RabbitMQ不在本機，只須要設置目標機器的IP地址或者機器名稱便可，而後設置前面建立的用戶名yy和密碼hello！。

緊接着要建立一個Channel，若是要發送消息，須要建立一個隊列，而後將消息發佈到這個隊列中。在建立隊列的時候，只有RabbitMQ上該隊列不存在，纔會去建立。消息是以二進制數組的形式傳輸的，因此若是消息是實體對象的話，須要序列化和而後轉化爲二進制數組。

如今客戶端發送代碼已經寫好了，運行以後，消息會發布到RabbitMQ的消息隊列中，如今須要編寫服務端的代碼鏈接到RabbitMQ上去獲取這些消息。

一樣，建立一個名爲Receive的服務端控制檯應用程序，服務端代碼以下：

static void Main(string[] args) { var factory = new ConnectionFactory(); factory.HostName = "localhost"; factory.UserName = "yy"; factory.Password = "hello!"; using (var connection = factory.CreateConnection()) { using (var channel = connection.CreateModel()) { channel.QueueDeclare("hello", false, false, false, null); var consumer = new QueueingBasicConsumer(channel); channel.BasicConsume("hello", true, consumer); Console.WriteLine(" waiting for message."); while (true) { var ea = (BasicDeliverEventArgs)consumer.Queue.Dequeue(); var body = ea.Body; var message = Encoding.UTF8.GetString(body); Console.WriteLine("Received {0}", message); } } } }

和發送同樣，首先須要定義鏈接，而後聲明消息隊列。要接收消息，須要定義一個Consume，而後從消息隊列中不斷Dequeue消息，而後處理。

如今發送端和接收端的代碼都寫好了，運行發送端，發送消息：

如今，名爲hello的消息隊列中，發送了一條消息。這條消息存儲到了RabbitMQ的服務器上了。使用rabbitmqctl 的list_queues能夠查看全部的消息隊列，以及裏面的消息個數，能夠看到，目前Rabbitmq上只有一個消息隊列，裏面只有一條消息：

D:\Program Files\RabbitMQ Server\rabbitmq_server-3.4.2\sbin>rabbitmqctl list_queues
Listing queues ...
hello   1

如今運行接收端程序，以下：

能夠看到，已經接受到了客戶端發送的Hello World，如今再來看RabitMQ上的消息隊列信息：

D:\Program Files\RabbitMQ Server\rabbitmq_server-3.4.2\sbin>rabbitmqctl list_queues
Listing queues ...
hello   0

能夠看到，hello這個隊列中的消息隊列個數爲0，這表示，當接收端，接收到消息以後，RabbitMQ上就把這個消息刪掉了。

2.2 工做隊列

前面的例子展現瞭如何往一個指定的消息隊列中發送和收取消息。如今咱們建立一個工做隊列（work queue）來將一些耗時的任務分發給多個工做者（workers）：

工做隊列（work queues, 又稱任務隊列Task Queues）的主要思想是爲了不當即執行並等待一些佔用大量資源、時間的操做完成。而是把任務（Task）看成消息發送到隊列中，稍後處理。一個運行在後臺的工做者（worker）進程就會取出任務而後處理。當運行多個工做者（workers）時，任務會在它們之間共享。

這個在網絡應用中很是有用，它能夠在短暫的HTTP請求中處理一些複雜的任務。在一些實時性要求不過高的地方，咱們能夠處理完主要操做以後，以消息的方式來處理其餘的不緊要的操做，好比寫日誌等等。

準備

在第一部分，發送了一個包含「Hello World!」的字符串消息。如今發送一些字符串，把這些字符串看成複雜的任務。這裏使用time.sleep()函數來模擬耗時的任務。在字符串中加上點號（.）來表示任務的複雜程度，一個點（.）將會耗時1秒鐘。好比"Hello..."就會耗時3秒鐘。

對以前示例的send.cs作些簡單的調整，以即可以發送隨意的消息。這個程序會按照計劃發送任務到咱們的工做隊列中。

static void Main(string[] args) { var factory = new ConnectionFactory(); factory.HostName = "localhost"; factory.UserName = "yy"; factory.Password = "hello!"; using (var connection = factory.CreateConnection()) { using (var channel = connection.CreateModel()) { channel.QueueDeclare("hello", false, false, false, null); string message = GetMessage(args);  var properties = channel.CreateBasicProperties(); properties.DeliveryMode = 2; var body = Encoding.UTF8.GetBytes(message); channel.BasicPublish("", "hello", properties, body); Console.WriteLine(" set {0}", message); } } Console.ReadKey(); } private static string GetMessage(string[] args) { return ((args.Length > 0) ? string.Join(" ", args) : "Hello World!"); }

加粗部分是通過修改過了的。

接着咱們修改接收端，讓他根據消息中的逗點的個數來Sleep對應的秒數：

static void Main(string[] args) { var factory = new ConnectionFactory(); factory.HostName = "localhost"; factory.UserName = "yy"; factory.Password = "hello!"; using (var connection = factory.CreateConnection()) { using (var channel = connection.CreateModel()) { channel.QueueDeclare("hello", false, false, false, null); var consumer = new QueueingBasicConsumer(channel); channel.BasicConsume("hello", true, consumer); while (true) { var ea = (BasicDeliverEventArgs)consumer.Queue.Dequeue(); var body = ea.Body; var message = Encoding.UTF8.GetString(body);  int dots = message.Split('.').Length - 1; Thread.Sleep(dots * 1000); Console.WriteLine("Received {0}", message); Console.WriteLine("Done"); } } } }

輪詢分發

使用工做隊列的一個好處就是它可以並行的處理隊列。若是堆積了不少任務，咱們只須要添加更多的工做者（workers）就能夠了，擴展很簡單。

如今，咱們先啓動兩個接收端，等待接受消息，而後啓動一個發送端開始發送消息。

在cmd條件下，發送了5條消息，每條消息後面的逗點表示該消息須要執行的時長，來模擬耗時的操做。

而後能夠看到，兩個接收端依次接收到了發出的消息：

默認，RabbitMQ會將每一個消息按照順序依次分發給下一個消費者。因此每一個消費者接收到的消息個數大體是平均的。這種消息分發的方式稱之爲輪詢（round-robin）。

2.3 消息響應

當處理一個比較耗時得任務的時候，也許想知道消費者（consumers）是否運行到一半就掛掉。在當前的代碼中，當RabbitMQ將消息發送給消費者（consumers）以後，立刻就會將該消息從隊列中移除。此時，若是把處理這個消息的工做者（worker）停掉，正在處理的這條消息就會丟失。同時，全部發送到這個工做者的尚未處理的消息都會丟失。

咱們不想丟失任何任務消息。若是一個工做者（worker）掛掉了，咱們但願該消息會從新發送給其餘的工做者（worker）。

爲了防止消息丟失，RabbitMQ提供了消息響應（acknowledgments）機制。消費者會經過一個ack（響應），告訴RabbitMQ已經收到並處理了某條消息，而後RabbitMQ纔會釋放並刪除這條消息。

若是消費者（consumer）掛掉了，沒有發送響應，RabbitMQ就會認爲消息沒有被徹底處理，而後從新發送給其餘消費者（consumer）。這樣，即便工做者（workers）偶爾的掛掉，也不會丟失消息。

消息是沒有超時這個概念的；當工做者與它斷開連的時候，RabbitMQ會從新發送消息。這樣在處理一個耗時很是長的消息任務的時候就不會出問題了。

消息響應默認是開啓的。在以前的例子中使用了no_ack=True標識把它關閉。是時候移除這個標識了，當工做者（worker）完成了任務，就發送一個響應。

channel.BasicConsume("hello", false, consumer); while (true) { var ea = (BasicDeliverEventArgs)consumer.Queue.Dequeue(); var body = ea.Body; var message = Encoding.UTF8.GetString(body); int dots = message.Split('.').Length - 1; Thread.Sleep(dots * 1000); Console.WriteLine("Received {0}", message); Console.WriteLine("Done"); channel.BasicAck(ea.DeliveryTag, false); }

如今,能夠保證,即便正在處理消息的工做者被停掉,這些消息也不會丟失,全部沒有被應答的消息會被從新發送給其餘工做者.

一個很常見的錯誤就是忘掉了BasicAck這個方法,這個錯誤很常見,可是後果很嚴重. 當客戶端退出時,待處理的消息就會被從新分發,可是RabitMQ會消耗愈來愈多的內存,由於這些沒有被應答的消息不可以被釋放。調試這種case，可使用rabbitmqct打印messages_unacknoledged字段。

rabbitmqctl list_queues name messages_ready messages_unacknowledged
Listing queues ...
hello    0       0
...done.

2.4 消息持久化

前面已經搞定了即便消費者down掉，任務也不會丟失，可是，若是RabbitMQ Server停掉了，那麼這些消息仍是會丟失。

當RabbitMQ Server 關閉或者崩潰，那麼裏面存儲的隊列和消息默認是不會保存下來的。若是要讓RabbitMQ保存住消息，須要在兩個地方同時設置：須要保證隊列和消息都是持久化的。

首先，要保證RabbitMQ不會丟失隊列，因此要作以下設置：

bool durable = true; channel.QueueDeclare("hello", durable, false, false, null);

雖然在語法上是正確的，可是在目前階段是不正確的，由於咱們以前已經定義了一個非持久化的hello隊列。RabbitMQ不容許咱們使用不一樣的參數從新定義一個已經存在的同名隊列，若是這樣作就會報錯。如今，定義另一個不一樣名稱的隊列：

bool durable = true; channel.queueDeclare("task_queue", durable, false, false, null);

queueDeclare 這個改動須要在發送端和接收端同時設置。

如今保證了task_queue這個消息隊列即便在RabbitMQ Server重啓以後，隊列也不會丟失。而後須要保證消息也是持久化的，這能夠經過設置IBasicProperties.SetPersistent 爲true來實現：

var properties = channel.CreateBasicProperties();
properties.SetPersistent(true);

須要注意的是，將消息設置爲持久化並不能徹底保證消息不丟失。雖然他告訴RabbitMQ將消息保存到磁盤上，可是在RabbitMQ接收到消息和將其保存到磁盤上這之間仍然有一個小的時間窗口。 RabbitMQ 可能只是將消息保存到了緩存中，並無將其寫入到磁盤上。持久化是不可以必定保證的，可是對於一個簡單任務隊列來講已經足夠。若是須要消息隊列持久化的強保證，可使用publisher confirms

2.5 公平分發

你可能會注意到，消息的分發可能並無如咱們想要的那樣公平分配。好比，對於兩個工做者。當奇數個消息的任務比較重，可是偶數個消息任務比較輕時，奇數個工做者始終處理忙碌狀態，而偶數個工做者始終處理空閒狀態。可是RabbitMQ並不知道這些，他仍然會平均依次的分發消息。

爲了改變這一狀態，咱們可使用basicQos方法，設置perfetchCount=1 。這樣就告訴RabbitMQ 不要在同一時間給一個工做者發送多於1個的消息，或者換句話說。在一個工做者還在處理消息，而且沒有響應消息以前，不要給他分發新的消息。相反，將這條新的消息發送給下一個不那麼忙碌的工做者。

channel.BasicQos(0, 1, false);

2.6 完整實例

如今將全部這些放在一塊兒：

發送端代碼以下：

static void Main(string[] args) { var factory = new ConnectionFactory(); factory.HostName = "localhost"; factory.UserName = "yy"; factory.Password = "hello!"; using (var connection = factory.CreateConnection()) { using (var channel = connection.CreateModel()) { bool durable = true; channel.QueueDeclare("task_queue", durable, false, false, null); string message = GetMessage(args); var properties = channel.CreateBasicProperties(); properties.SetPersistent(true); var body = Encoding.UTF8.GetBytes(message); channel.BasicPublish("", "task_queue", properties, body); Console.WriteLine(" set {0}", message); } } Console.ReadKey(); } private static string GetMessage(string[] args) { return ((args.Length > 0) ? string.Join(" ", args) : "Hello World!"); }

接收端代碼以下：

static void Main(string[] args) { var factory = new ConnectionFactory(); factory.HostName = "localhost"; factory.UserName = "yy"; factory.Password = "hello!"; using (var connection = factory.CreateConnection()) { using (var channel = connection.CreateModel()) { bool durable = true; channel.QueueDeclare("task_queue", durable, false, false, null); channel.BasicQos(0, 1, false); var consumer = new QueueingBasicConsumer(channel); channel.BasicConsume("task_queue", false, consumer); while (true) { var ea = (BasicDeliverEventArgs)consumer.Queue.Dequeue(); var body = ea.Body; var message = Encoding.UTF8.GetString(body); int dots = message.Split('.').Length - 1; Thread.Sleep(dots * 1000); Console.WriteLine("Received {0}", message); Console.WriteLine("Done"); channel.BasicAck(ea.DeliveryTag, false); } } } }

三管理界面

RabbitMQ還有一個管理界面，經過該界面能夠查看RabbitMQ Server 當前的狀態，該界面是以插件形式提供的，而且在安裝RabbitMQ的時候已經自帶了該插件。須要作的是在RabbitMQ控制檯界面中啓用該插件，命令以下：

rabbitmq-plugins enable rabbitmq_management

如今，在瀏覽器中輸入 http://server-name:15672/ server-name換成機器地址或者域名，若是是本地的，直接用localhost（RabbitMQ 3.0以前版本端口號爲55672）在輸入以後，彈出登陸界面，使用咱們以前建立的用戶登陸。

在該界面上能夠看到當前RabbitMQServer的全部狀態。

四總結

本文簡單介紹了消息隊列的相關概念，並介紹了RabbitMQ消息代理的基本原理以及在Windows 上如何安裝RabbitMQ和在.NET中如何使用RabbitMQ。消息隊列在構建分佈式系統和提升系統的可擴展性和響應性方面有着很重要的做用，但願本文對您瞭解消息隊列以及如何使用RabbitMQ有所幫助。

五參考文獻

幾個概念：

Exchange:交換機，決定了消息路由規則；

Queue:消息隊列；

Channel:進行消息讀寫的通道；

Bind:綁定了Queue和Exchange，意即爲符合什麼樣路由規則的消息，將會放置入哪個消息隊列；

RabbitMQ是流行的開源消息隊列系統，用erlang語言開發。我曾經對這門語言挺有興趣，學過一段時間，後來沒堅持。RabbitMQ是 AMQP（高級消息隊列協議）的標準實現。若是不熟悉AMQP，直接看RabbitMQ的文檔會比較困難。不過它也只有幾個關鍵概念，這裏簡單介紹。[1]

RabbitMQ的結構圖以下：

幾個概念說明：

Broker：簡單來講就是消息隊列服務器實體。　　Exchange：消息交換機，它指定消息按什麼規則，路由到哪一個隊列。　　Queue：消息隊列載體，每一個消息都會被投入到一個或多個隊列。　　Binding：綁定，它的做用就是把exchange和queue按照路由規則綁定起來。　　Routing Key：路由關鍵字，exchange根據這個關鍵字進行消息投遞。　　vhost：虛擬主機，一個broker裏能夠開設多個vhost，用做不一樣用戶的權限分離。　　producer：消息生產者，就是投遞消息的程序。　　consumer：消息消費者，就是接受消息的程序。　　channel：消息通道，在客戶端的每一個鏈接裏，可創建多個channel，每一個channel表明一個會話任務。

消息隊列的使用過程大概以下：

（1）客戶端鏈接到消息隊列服務器，打開一個channel。　　（2）客戶端聲明一個exchange，並設置相關屬性。　　（3）客戶端聲明一個queue，並設置相關屬性。　　（4）客戶端使用routing key，在exchange和queue之間創建好綁定關係。　　（5）客戶端投遞消息到exchange。

exchange接收到消息後，就根據消息的key和已經設置的binding，進行消息路由，將消息投遞到一個或多個隊列裏。

exchange也有幾個類型，徹底根據key進行投遞的叫作Direct交換機，例如，綁定時設置了routing key爲」abc」，那麼客戶端提交的消息，只有設置了key爲」abc」的纔會投遞到隊列。對key進行模式匹配後進行投遞的叫作Topic交換機，符號」#」匹配一個或多個詞，符號」*」匹配正好一個詞。例如」abc.#」匹配」abc.def.ghi」，」abc.*」只匹配」abc.def」。還有一種不須要key的，叫作Fanout交換機，它採起廣播模式，一個消息進來時，投遞到與該交換機綁定的全部隊列。

RabbitMQ支持消息的持久化，也就是數據寫在磁盤上，爲了數據安全考慮，我想大多數用戶都會選擇持久化。消息隊列持久化包括3個部分：　　（1）exchange持久化，在聲明時指定durable => 1 　　（2）queue持久化，在聲明時指定durable => 1 　　（3）消息持久化，在投遞時指定delivery_mode => 2（1是非持久化）

若是exchange和queue都是持久化的，那麼它們之間的binding也是持久化的。若是exchange和queue二者之間有一個持久化，一個非持久化，就不容許創建綁定。[1]

參考： http://baike.baidu.com/view/4095865.htm?fr=aladdin