NET 環境中使用RabbitMQ

来源:http://www.cnblogs.com/wangchaozhi/archive/2016/01/28/5161560.html
-Advertisement-
Play Games

一 環境搭建 首先,由於RabbitMQ使用Erlang編寫的,需要運行在Erlang運行時環境上,所以在安裝RabbitMQ Server之前需要安裝Erlang 運行時環境,可以到Erlang官網下載對應平臺的安裝文件。如果沒有安裝運行時環境,安裝RabbitMQ Server的時候,會提示需要


 

一 環境搭建

首先,由於RabbitMQ使用Erlang編寫的,需要運行在Erlang運行時環境上,所以在安裝RabbitMQ Server之前需要安裝Erlang 運行時環境,可以到Erlang官網下載對應平臺的安裝文件。如果沒有安裝運行時環境,安裝RabbitMQ Server的時候,會提示需要先安裝Erlang環境。 安裝完成之後,確保已經將Erlang的安裝路徑註冊到系統的環境變數中。安裝完Erlang之後,這個環境會自動設置,如果沒有:按照下圖進行設置。

 

然後,去RabbitMQ官網下載RabbitMQ Server服務端程式,選擇合適的平臺版本下載。安裝完成之後,就可以開始使用了。

現在就可以對RabbitMQ Server進行配置了。

首先,切換到RabbitMQ Server的安裝目錄:

在sbin下麵有很多batch文件,用來控制RabbitMQ Server。

最簡單的方式是使RabbitMQ以Windows Service的方式在後臺運行,所以我們需要以管理員許可權打開cmd,然後切換到sbin目錄下,執行這三條命令即可:

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

現在RabbitMQ的服務端已經啟動起來了(如果啟動失敗,請檢查安裝後是否服務已經啟動,如果沒有,則可能是因為安裝版本的原因)。

下麵可以使用sbin目錄下麵的rabbitmqctl.bat這個腳本來查看和控制服務端狀態的,在cmd中直接運行rabbitmqctl status。如果不是看到以下結果:,需要到C:\Windows目錄下,將.erlang.cookie文件,拷貝到用戶目錄下 C:\Users\{用戶名},這是Erlang的Cookie文件,允許與Erlang進行交互:

RabbitMQ Server上面也有用戶概念,安裝好之後,使用rabbitmqctl list_users命令,可以看到上面目前的用戶:

可以使用下麵的命令來添加用戶並設置許可權:

rabbitmqctl  add_user  test  123456
rabbitmqctl  set_permissions  test  ".*"  ".*"  ".*"
rabbitmqctl  set_user_tags test administrator

上面的一條命令添加了一個名為test的用戶,並設置了密碼123456,下麵的命令為用戶test分別授予對所有消息隊列的配置、讀和寫的許可權。

現在我們可以將預設的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"來表示:

producer

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

queue

  • 消費(Consuming)和獲取消息是一樣的意思。一個消費者(consumer)就是一個等待獲取消息的程式。我們把它畫作"C":

consumer

通常,消息生產者,消息消費者和消息代理不在同一臺機器上。

2.1 Hello World

為了展示RabbitMQ的基本使用,我們發送一個HelloWorld消息,然後接收並處理。

rabbitmq hello world

首先創建一個控制台程式,用來將消息發送到RabbitMQ的消息隊列中,代碼如下:


首先,需要創建一個ConnectionFactory,設置目標,由於是在本機,所以設置為localhost,如果RabbitMQ不在本機,只需要設置目標機器的IP地址或者機器名稱即可,然後設置前面創建的用戶名test和密碼123456。

緊接著要創建一個Channel,如果要發送消息,需要創建一個隊列,然後將消息發佈到這個隊列中。在創建隊列的時候,只有RabbitMQ上該隊列不存在,才會去創建。消息是以二進位數組的形式傳輸的,所以如果消息是實體對象的話,需要序列化和然後轉化為二進位數組。

現在客戶端發送代碼已經寫好了,運行之後,消息會發佈到RabbitMQ的消息隊列中,現在需要編寫服務端的代碼連接到RabbitMQ上去獲取這些消息。

同樣,創建一個名為Receive的服務端控制台應用程式,服務端代碼如下:


和發送一樣,首先需要定義連接,然後聲明消息隊列。要接收消息,需要定義一個Consume,然後從消息隊列中不斷Dequeue消息,然後處理。

現在發送端和接收端的代碼都寫好了,運行發送端,發送消息:

 

現在,名為hello的消息隊列中,發送了一條消息。這條消息存儲到了RabbitMQ的伺服器上了。使用rabbitmqctl 的list_queues可以查看所有的消息隊列,以及裡面的消息個數,可以看到,目前Rabbitmq上只有一個消息隊列,裡面只有一條消息:

rabbitmqctl list_queues
Listing queues ...
hello   1

現在運行接收端程式:

可以看到,已經接受到了客戶端發送的Hello World,現在再來看RabitMQ上的消息隊列信息:

rabbitmqctl list_queues
Listing queues ...
hello   0

可以看到,hello這個隊列中的消息隊列個數為0,這表示,當接收端,接收到消息之後,RabbitMQ上就把這個消息刪掉了。

2.2 工作隊列

前面的例子展示瞭如何往一個指定的消息隊列中發送和收取消息。現在我們創建一個工作隊列(work queue)來將一些耗時的任務分發給多個工作者(workers):

rabbitmq-work queue

工作隊列(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 = "test";
    factory.Password = "123456";

    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 = "test";
    factory.Password = "123456";

    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)就可以了,擴展很簡單。

現在,我們先啟動兩個接收端,等待接受消息,然後啟動一個發送端開始發送消息。

Send message queue 

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

然後可以看到,兩個接收端依次接收到了發出的消息:

receive message queue 

預設,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 = "test";
    factory.Password = "123456";

    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 = "test";
    factory.Password = "123456";

    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

rabbitmq management

現在,在瀏覽器中輸入 http://server-name:15672/ server-name換成機器地址或者功能變數名稱,如果是本地的,直接用localhost在輸入之後,彈出登錄界面,使用我們之前創建的用戶登錄。

 .

在該界面上可以看到當前RabbitMQServer的所有狀態。

四 總結

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

五 參考文獻

  1. http://www.infoq.com/cn/articles/message-based-distributed-architecture
  2. http://www.rabbitmq.com/getstarted.html
  3. http://www.codethinked.com/using-rabbitmq-with-c-and-net
  4. http://www.infoq.com/cn/articles/AMQP-RabbitMQ
  5. http://www.infoq.com/cn/articles/ebay-scalability-best-practices

您的分享是我們最大的動力!

-Advertisement-
Play Games
更多相關文章
  • 一列數的規則如下: 1、1、2、3、5、8、13、21、34...... 求第30位數是多少, 用遞歸演算法實現。 代碼: public class MainClass { public static void Main() { Console.WriteLine(Foo(30)); } public
  • 在寫一個客戶的B/S結構應用程式時,突然發現一個技巧,不知道是否是MS的一個BUG,給相關的有研究的朋友原先考慮寫一個檢查Session的類,Session失效後,必須轉向登陸頁面,可每一個調用該類的頁面,在不同的WEB路徑下,所以轉到登陸頁面的URL都不同,每個頁面都要調用和設置登陸頁面路徑,所以
  • 在TextBox控制項中使用快捷鍵,一般要求按下快捷鍵立刻產生效果,KeyUp事件顯然不符合我們的要求,而KeyPress事件中不支持使用組合件,所以我們選用KeyDown事件,具體代碼實現如下: private void tBBefore_KeyDown(object sender, KeyEven...
  • 一、asp:Repeater <div class="bd"> <ul> <asp:Repeater ID="rept_slide" runat="server"> <ItemTemplate> <li><a href='<%#Eval("LinkUrl").ToString() %>' targe
  • 1.0 創建Attribute using System; using System.Collections.Generic; using System.Linq; using System.Web; namespace LSUnion.Site.WebHelper { [AttributeUsag
  • .NET之死是JAVA引起的嗎?.NET為什麼會死?.NET之死預示著什麼?
  • 一.基礎篇 C#不像C++,他本身是沒有聯合Union的,但是可以通過手動控制結構體每個元素的位置來實現,這需要結合使用StructLayoutAttribute、LayoutKind以及FieldOffsetAttribute。使用它們的時候必須引用System.Runtime.InteropSe
  • 自ExcelUtility類推出以來,經過項目中的實際使用與不斷完善,現在又做了許多的優化並增加了許多的功能,本篇不再講述原理,直接貼出示例代碼以及相關的模板、結果圖,以便大家快速掌握,另外這些示例說明我也已同步到GIT中,大家可以下載與學習,不足之處,敬請見諒,謝謝! 一、ExcelUtility
一周排行
    -Advertisement-
    Play Games
  • 移動開發(一):使用.NET MAUI開發第一個安卓APP 對於工作多年的C#程式員來說,近來想嘗試開發一款安卓APP,考慮了很久最終選擇使用.NET MAUI這個微軟官方的框架來嘗試體驗開發安卓APP,畢竟是使用Visual Studio開發工具,使用起來也比較的順手,結合微軟官方的教程進行了安卓 ...
  • 前言 QuestPDF 是一個開源 .NET 庫,用於生成 PDF 文檔。使用了C# Fluent API方式可簡化開發、減少錯誤並提高工作效率。利用它可以輕鬆生成 PDF 報告、發票、導出文件等。 項目介紹 QuestPDF 是一個革命性的開源 .NET 庫,它徹底改變了我們生成 PDF 文檔的方 ...
  • 項目地址 項目後端地址: https://github.com/ZyPLJ/ZYTteeHole 項目前端頁面地址: ZyPLJ/TreeHoleVue (github.com) https://github.com/ZyPLJ/TreeHoleVue 目前項目測試訪問地址: http://tree ...
  • 話不多說,直接開乾 一.下載 1.官方鏈接下載: https://www.microsoft.com/zh-cn/sql-server/sql-server-downloads 2.在下載目錄中找到下麵這個小的安裝包 SQL2022-SSEI-Dev.exe,運行開始下載SQL server; 二. ...
  • 前言 隨著物聯網(IoT)技術的迅猛發展,MQTT(消息隊列遙測傳輸)協議憑藉其輕量級和高效性,已成為眾多物聯網應用的首選通信標準。 MQTTnet 作為一個高性能的 .NET 開源庫,為 .NET 平臺上的 MQTT 客戶端與伺服器開發提供了強大的支持。 本文將全面介紹 MQTTnet 的核心功能 ...
  • Serilog支持多種接收器用於日誌存儲,增強器用於添加屬性,LogContext管理動態屬性,支持多種輸出格式包括純文本、JSON及ExpressionTemplate。還提供了自定義格式化選項,適用於不同需求。 ...
  • 目錄簡介獲取 HTML 文檔解析 HTML 文檔測試參考文章 簡介 動態內容網站使用 JavaScript 腳本動態檢索和渲染數據,爬取信息時需要模擬瀏覽器行為,否則獲取到的源碼基本是空的。 本文使用的爬取步驟如下: 使用 Selenium 獲取渲染後的 HTML 文檔 使用 HtmlAgility ...
  • 1.前言 什麼是熱更新 游戲或者軟體更新時,無需重新下載客戶端進行安裝,而是在應用程式啟動的情況下,在內部進行資源或者代碼更新 Unity目前常用熱更新解決方案 HybridCLR,Xlua,ILRuntime等 Unity目前常用資源管理解決方案 AssetBundles,Addressable, ...
  • 本文章主要是在C# ASP.NET Core Web API框架實現向手機發送驗證碼簡訊功能。這裡我選擇是一個互億無線簡訊驗證碼平臺,其實像阿裡雲,騰訊雲上面也可以。 首先我們先去 互億無線 https://www.ihuyi.com/api/sms.html 去註冊一個賬號 註冊完成賬號後,它會送 ...
  • 通過以下方式可以高效,並保證數據同步的可靠性 1.API設計 使用RESTful設計,確保API端點明確,並使用適當的HTTP方法(如POST用於創建,PUT用於更新)。 設計清晰的請求和響應模型,以確保客戶端能夠理解預期格式。 2.數據驗證 在伺服器端進行嚴格的數據驗證,確保接收到的數據符合預期格 ...