C#調(diào)用RabbitMQ實現(xiàn)消息隊列的示例代碼

前言

我在剛接觸使用中間件的時候，發(fā)現(xiàn)，中間件的使用并不是最難的，反而是中間件的下載，安裝，配置才是最難的。

所以，這篇文章我們從頭開始學(xué)習(xí)RabbitMq，真正的從頭開始。

關(guān)于消息隊列

其實消息隊列沒有那么神秘，我們這樣想一下，用戶訪問網(wǎng)站，最終是要將數(shù)據(jù)以HTTP的協(xié)議的方式，通過網(wǎng)絡(luò)傳輸?shù)街鳈C的某個端口上的。

那么，接收數(shù)據(jù)的方式是什么呢？自然是端口監(jiān)聽啦。

那消息隊列是什么就很好解釋了？

它就是端口監(jiān)聽，接到數(shù)據(jù)后，將數(shù)據(jù)排列起來。

那這件事，我們不用中間件能做嗎？

當(dāng)然能做啦，寫個TCP/UDP/Socket的軟件就可以做啦。

舉個簡單的例子，如下圖：

既然自己可以做消息隊列，那為什么要用RabbitMQ?

因為，RabbitMQ成熟的開源中間件，可靠性有保證，bug少，性能也非常好。

而C#代碼默認是使用托管內(nèi)存的，所以，想寫出媲美RabbitMQ性能的消息隊列，就必須離開我們常用的托管內(nèi)存，使用非托管內(nèi)存，但這個代價就太大了；而且最終能否達到RabbitMQ的性能水平還是個未知數(shù)。

還有就是RabbitMQ除了基礎(chǔ)的消息隊列管理，還有很多很強大的額外功能，而自己開發(fā)消息隊列，很難如此盡善盡美。

我們還會發(fā)現(xiàn)，在消息隊列里有很多概念，什么消息總線啊，什么工作隊列啊等等。

要怎么理解這些概念呢？

很簡單，不要去理解。這些概念其實是人家代碼架構(gòu)的模式，不要去理解他們，【記】就完了，人家的中間件就是按照這個模式工作的。

比如，我寫了一個接收消息的總控制器，然后我為他命名為總線，那這個控制器就是總線，沒有理由，這就是定義。

準(zhǔn)備工作

首先，我們訪問官網(wǎng)【https://www.rabbitmq.com/】，點擊Get Started。

然后，網(wǎng)站會自動跳轉(zhuǎn)到當(dāng)前首頁Get Started的錨點位置，如下圖：

Get Started錨點：

然后我們點擊DownLoad+Installation，進入到下載界面。

在下載頁面中，我們找到安裝指南，然后在點擊官網(wǎng)推薦的Windows系統(tǒng)的安裝包，如下圖：

現(xiàn)在，我們進入了Windows安裝指南界面了。

首先，我們看一下預(yù)覽信息，如下圖：

在預(yù)覽里，我們得知，安裝RabbitMQ有兩種方法，一種是使用Chocolatey安裝，一種是使用官方安裝包安裝。

Chocolatey是什么呢?隨手百度一下，原來他是一個軟件包管理工具，也就是說，Chocolatey是類似于Nuget的一種工具。

由于Chocolatey的使用，我不是很熟悉，所以，這里選擇使用官方安裝包安裝。

點擊【Using the official installer】，我們進入了【Using the official installer】對應(yīng)的錨點，如下圖。

在【Using the official installer】段落里找到有推薦標(biāo)志的安裝包，然后下載。

下載完成后，我們可以得到這樣一個安裝包，如下圖：

除了下載安裝包，我們還會發(fā)現(xiàn)，在【Using the official installer】段落里，有提醒我們，RabbitMQ是有依賴的，依賴一個Erlang語言的框架（類似于C#語言的NetFramework）。

我們可以發(fā)現(xiàn)，在依賴的段落里，官網(wǎng)非常坑的給出了三個鏈接網(wǎng)址，如下：

supported version of Erlang：https://www.rabbitmq.com/which-erlang.html

Windows installer：https://www.erlang.org/downloads

Erlang Solutions：https://www.erlang-solutions.com/resources/download.html

因為，我們是無法通過文字描述來判斷，哪一個是真的依賴框架的下載地址，所以只好每個都點擊進去看看。。。

打開網(wǎng)址后發(fā)現(xiàn)，在后兩個網(wǎng)址中都可以找到框架下載地址，但第二個地址明顯更友好一點，所以我們在第二個網(wǎng)址內(nèi)下載Erlang的框架。

下載完成得到如下圖文件：

PS：這里下載的是OTP的22.1的版本，我的理解是Erlang等于C#語言，而OTP等于NetFramework。

安裝Erlang\OTP

首先，我們運行otp_win64_22.1.exe，安裝依賴框架Erlang\OTP。

安裝完成后，設(shè)置環(huán)境變量如下：

然后運行CMD，輸入erl，測試安裝是否成功，如下圖：

安裝成功。

安裝rabbitmq-server

安裝完依賴后，我們接著安裝rabbitmq-server-3.8.0.exe。

【rabbitmq-server-3.8.0.exe】？從這個文件名上，我們發(fā)現(xiàn)了一個問題，那就是，我們即將安裝的RabbitMQ，是一個服務(wù)端啊。

什么？服務(wù)端？難道還有客戶端？？？

其實這也很好理解，想一下最開始我舉的那個例子，消息隊列是需要一個監(jiān)聽端口的服務(wù)端的，然后客戶端向這個服務(wù)端發(fā)送請求。

這樣是不是就很好的理解RabbitMQ了呢：）

安裝完RabbitMQ服務(wù)端后，我們還是啟動CMD，用命令行來查看下安裝狀態(tài)。

首先輸入下面的命令，將路徑定位到RabbitMQ的路徑下：

【CD /D C:\Program Files\RabbitMQ Server\rabbitmq_server-3.8.0\sbin】

然后輸入rabbitmqctl status查看狀態(tài)。

啟動管理工具的命令行：rabbitmq-plugins enable rabbitmq_management。

啟動成功后，在瀏覽器輸入地址http://127.0.0.1:15672/，進入管理頁面，賬戶密碼都是guest。

RabbitMQ還有很多常用命令，大家可以自行百度。

到此，RabbitMQ服務(wù)端的環(huán)境配置好了，正常情況，這些配置應(yīng)該在服務(wù)器進行，但我為了測試方便，就把服務(wù)端也安裝在本機了，因此我下面調(diào)用RabbitMQ時，連接的主機IP都是localhost。

RabbitMQ應(yīng)用

首先創(chuàng)建兩個控制臺應(yīng)用程序，KibaRabbitMQSend和KibaRabbitMQReceived。

然后引入RabbitMQ的開源類庫。

在C#里使用RabbitMQ開源類庫非常簡單，可以去官網(wǎng)下載一個.NET版本的RabbitMQ客戶端類庫，也可以直接在Nuget上搜索RabbitMQ，然后安裝，如下圖：

KibaRabbitMQSend

安裝完RabbitMQ開源類庫后，我們編寫代碼，實現(xiàn)向RabbitMQ服務(wù)器發(fā)送消息，代碼如下：

static void Main(string[] args)
{
  var factory = new ConnectionFactory();
  factory.HostName = "localhost";//主機名，Rabbit會拿這個IP生成一個endpoint，這個很熟悉吧，就是socket綁定的那個終結(jié)點。
  factory.UserName = "guest";//默認用戶名,用戶可以在服務(wù)端自定義創(chuàng)建，有相關(guān)命令行
  factory.Password = "guest";//默認密碼
 
  using (var connection = factory.CreateConnection())//連接服務(wù)器，即正在創(chuàng)建終結(jié)點。
  {
    //創(chuàng)建一個通道，這個就是Rabbit自己定義的規(guī)則了，如果自己寫消息隊列，這個就可以開腦洞設(shè)計了
    //這里Rabbit的玩法就是一個通道channel下包含多個隊列Queue
    using (var channel = connection.CreateModel())
    {
       channel.QueueDeclare("kibaQueue", false, false, false, null);//創(chuàng)建一個名稱為kibaqueue的消息隊列
       var properties = channel.CreateBasicProperties();
       properties.DeliveryMode = 1;
       string message = "I am Kiba518"; //傳遞的消息內(nèi)容
       channel.BasicPublish("", "kibaQueue", properties, Encoding.UTF8.GetBytes(message)); //生產(chǎn)消息
       Console.WriteLine($"Send:{message}");
    }
  }
}

運行代碼。

然后我們使用命令行rabbitmqctl list_queues，去RabbitMQ的服務(wù)器查看當(dāng)前消息隊列，如下圖：

可以看到，我們的消息已經(jīng)發(fā)送成功了。

KibaRabbitMQReceived

現(xiàn)在我們編寫接收消息代碼，如下：

static void Main(string[] args)
{
  var factory = new ConnectionFactory();
  factory.HostName = "localhost";
  factory.UserName = "guest";
  factory.Password = "guest";
 
  using (var connection = factory.CreateConnection())
  {
    using (var channel = connection.CreateModel())
    {
      channel.QueueDeclare("kibaQueue", false, false, false, null);
 
      /* 這里定義了一個消費者，用于消費服務(wù)器接受的消息
       * C#開發(fā)需要注意下這里，在一些非面向?qū)ο蠛兔嫦驅(qū)ο蟊容^差的語言中，是非常重視這種設(shè)計模式的。
       * 比如RabbitMQ使用了生產(chǎn)者與消費者模式，然后很多相關(guān)的使用文章都在拿這個生產(chǎn)者和消費者來表述。
       * 但是，在C#里，生產(chǎn)者與消費者對我們而言，根本算不上一種設(shè)計模式，他就是一種最基礎(chǔ)的代碼編寫規(guī)則。
       * 所以，大家不要復(fù)雜的名詞嚇到，其實，并沒那么復(fù)雜。
       * 這里，其實就是定義一個EventingBasicConsumer類型的對象，然后該對象有個Received事件，
       * 該事件會在服務(wù)接收到數(shù)據(jù)時觸發(fā)。
       */
      var consumer = new EventingBasicConsumer(channel);//消費者
      channel.BasicConsume("kibaQueue", true, consumer);//消費消息
      consumer.Received += (model, ea) =>
      {
        var body = ea.Body;
        var message = Encoding.UTF8.GetString(body);
      };
    }
  }
}

運行代碼。

然后我們使用命令行rabbitmqctl list_queues，去RabbitMQ的服務(wù)器查看當(dāng)前消息隊列，如下圖：

可以看到，消息已經(jīng)被使用了。

現(xiàn)在我們在發(fā)送代碼出做一個for循環(huán)，看看消息接收速度是什么樣的，代碼如下，for循環(huán)了100次，每次間隔3秒。

for (int i = 0; i < 100; i++)
{
  channel.QueueDeclare("kibaQueue", false, false, false, null);//創(chuàng)建一個名稱為kibaQueue的消息隊列
  var properties = channel.CreateBasicProperties();
  properties.DeliveryMode = 1;
  string message = "I am Kiba518"; //傳遞的消息內(nèi)容
  channel.BasicPublish("", "kibaQueue", properties, Encoding.UTF8.GetBytes(message)); //生產(chǎn)消息
  Console.WriteLine($"Send:{message}");
  Thread.Sleep(3000);
}

效果圖如下：

可以看到，發(fā)送消息和接收消息，幾乎是同步的，效果非常理想。

服務(wù)器端應(yīng)用

在上文，我們的RabbitMQ服務(wù)是安裝在我的本機上的；現(xiàn)在我們把服務(wù)移植到服務(wù)器上，然后再來測試一下。

在服務(wù)器端安裝RabbitMQ和在本機安裝的步驟是一樣的，但是安裝完成后，我們需要設(shè)置下防火墻的入站規(guī)則和出站規(guī)則，將5672的UDP端口開放一下。

為什么要開放端口是5672？因為RabbitMQ的默認的消息接收和發(fā)送端口就是5672，我們可以使用斷點查看一下。

如上圖，可以看到，在我們沒有設(shè)置端口的時候，Endpoint的端口的默認值是5672。

配置完端口后，我們修改代碼中的HostName為我們的服務(wù)器地址，如下。

factory.HostName = "1.1.1.1";

重新運行代碼，會發(fā)現(xiàn)在運行到factory.CreateConnection()的時候，系統(tǒng)提示一個異?！綬abbitMQ.Client.Exceptions.BrokerUnreachableException:“None of the specified endpoints were reachable”】，如下圖：

這是因為我們使用的賬號是guest，guest賬號默認是不支持遠程連接的。

解決辦法很簡單，新建一個賬戶即可。

創(chuàng)建用戶

在服務(wù)器端打開瀏覽器，輸入http://127.0.0.1:15672/，進入管理頁面。

點擊菜單欄的Admin選項，進入用戶管理界面創(chuàng)建用戶kiba，密碼123456，如下圖：

創(chuàng)建完用戶后，得到如下界面。

如上圖所示，剛剛創(chuàng)建的用戶還沒有任何訪問權(quán)限。

現(xiàn)在我們點擊用戶名，進入權(quán)限管理頁面設(shè)置權(quán)限。

如上圖所示，頁面默認為我們設(shè)置了一個可讀，可寫，可管理配置的權(quán)限；所以，我們只要點擊Setpremission就可以了。

設(shè)置完權(quán)限，我們回到用戶管理頁面。

如上圖所示，權(quán)限設(shè)置成功。

現(xiàn)在我們回到代碼，修改用戶名密碼如下。

factory.HostName = "1.1.1.1";
factory.UserName = "kiba";
factory.Password = "123456";

運行代碼，不再拋異常，接受發(fā)送消息正常。

設(shè)置用戶權(quán)限也可以通過命令的方式設(shè)置，如下：

rabbitmqctl set_permissions -p "/" kiba "." "." ".*"

到此C#調(diào)用RabbitMQ實現(xiàn)消息隊列就講完了。

代碼已經(jīng)傳到Github上了，歡迎大家下載。

Github地址：https://github.com/kiba518/KibaRabbitMQ

以上就是本文的全部內(nèi)容，希望對大家的學(xué)習(xí)有所幫助，也希望大家多多支持億速云。