1 Environment setup First of all, since RabbitMQ is written in Erlang and needs to run on the Erlang runtime environment, you need to install the Erlang runtime environment before installing RabbitMQ Server. You can download the installation file for the corresponding platform from the Erlang official website. If the runtime environment is not installed, when installing RabbitMQ Server, you will be prompted to install it first

1 Environment setup

First of all, since RabbitMQ is written in Erlang and needs to run on the Erlang runtime environment, you need to install the Erlang runtime environment before installing RabbitMQ Server. You can download the installation file for the corresponding platform from the Erlang official website. If the runtime environment is not installed, when installing RabbitMQ Server, you will be prompted to install the Erlang environment first. After the installation is complete, make sure that the Erlang installation path has been registered in the system's environment variables. After installing Erlang, this environment will be set up automatically. If not: set it up as shown below.

Then, go to the RabbitMQ official website to download the RabbitMQ Server server program, and select the appropriate platform version to download. After the installation is complete, you can start using it.

Now you can configure RabbitMQ Server.

First, switch to the installation directory of RabbitMQ Server:

There are many batch files under sbin, used to control RabbitMQ Server.

The simplest way is to make RabbitMQ run in the background as a Windows Service, so we need to open cmd with administrator rights, then switch to the sbin directory and execute these three commands:

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

Now the RabbitMQ server has been started (If the startup fails, please check whether the service has been started after installation. If not, it may be because of the installed version).

You can use the rabbitmqctl.bat script under the sbin directory to view and control the server status. Run rabbitmqctl status directly in cmd. If you do not see the following result: You need to go to the C:Windows directory and copy the .erlang.cookie file to the user directory C:Users{username}. This is the Erlang cookie file, allowing interaction with Erlang:

RabbitMQ Server also has the concept of users. After installation, use the rabbitmqctl list_users command to see the current users above:

You can use the following commands to add users and set permissions:

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

The above command adds a user named test and sets the password 123456. The following command grants user test the configuration, read and write permissions for all message NET environment using RabbitMQs.

Now we can delete the default guest user by using the following command:

rabbitmqctl delete_user guest

If you want to change the password, you can use the following command:

rabbitmqctl change_passWord {username}  {newpassowrd}

2 Get started

To use RabbitMQ in .NET, you need to download the RabbitMQ client assembly. You can download it from the official website. After downloading and decompressing, you can get RabbitMQ.Client.dll, which is the RabbitMQ client.

Before using RabitMQ, you need to explain the following basic concepts:

RabbitMQ is a message broker. It receives messages from message NET environment using RabbitMQs (PRoducers), and then sends the messages to message NET environment using RabbitMQs (NET environment using RabbitMQs). Between sending and receiving, it can route, cache and persist according to the set rules.

Generally speaking, some proper nouns are used when referring to RabbitMQ and messages.

• Producing means sending. The program that sends the message is a NET environment using RabbitMQ. We usually use "P" to represent:

• Queue is the name of the mailbox. Messages are transferred between your application and RabbitMQ, and they can only be stored in NET environment using RabbitMQs. There is no limit to the capacity of the NET environment using RabbitMQ, you can store as many messages as you want - basically an infinite buffer. Multiple NET environment using RabbitMQs can send messages to the same NET environment using RabbitMQ, and multiple NET environment using RabbitMQs can also obtain data from the same NET environment using RabbitMQ. The NET environment using RabbitMQ can be drawn like this (the picture is the name of the NET environment using RabbitMQ):

• 消費(fèi)（Consuming）和獲取消息是一樣的意思。一個(gè)消費(fèi)者（NET environment using RabbitMQ）就是一個(gè)等待獲取消息的程序。我們把它畫作"C"：

通常，消息生產(chǎn)者，消息消費(fèi)者和消息代理不在同一臺(tái)機(jī)器上。

2.1 Hello World

為了展示RabbitMQ的基本使用，我們發(fā)送一個(gè)HelloWorld消息，然后接收并處理。

首先創(chuàng)建一個(gè)控制臺(tái)程序，用來(lái)將消息發(fā)送到RabbitMQ的消息隊(duì)列中，代碼如下：

首先，需要?jiǎng)?chuàng)建一個(gè)ConnectionFactory，設(shè)置目標(biāo)，由于是在本機(jī)，所以設(shè)置為localhost，如果RabbitMQ不在本機(jī)，只需要設(shè)置目標(biāo)機(jī)器的ip地址或者機(jī)器名稱即可，然后設(shè)置前面創(chuàng)建的用戶名test和密碼123456。

緊接著要?jiǎng)?chuàng)建一個(gè)Channel，如果要發(fā)送消息，需要?jiǎng)?chuàng)建一個(gè)隊(duì)列，然后將消息發(fā)布到這個(gè)隊(duì)列中。在創(chuàng)建隊(duì)列的時(shí)候，只有RabbitMQ上該隊(duì)列不存在，才會(huì)去創(chuàng)建。消息是以二進(jìn)制數(shù)組的形式傳輸?shù)?，所以如果消息是?shí)體對(duì)象的話，需要序列化和然后轉(zhuǎn)化為二進(jìn)制數(shù)組。

現(xiàn)在客戶端發(fā)送代碼已經(jīng)寫好了，運(yùn)行之后，消息會(huì)發(fā)布到RabbitMQ的消息隊(duì)列中，現(xiàn)在需要編寫服務(wù)端的代碼連接到RabbitMQ上去獲取這些消息。

同樣，創(chuàng)建一個(gè)名為Receive的服務(wù)端控制臺(tái)應(yīng)用程序，服務(wù)端代碼如下：

和發(fā)送一樣，首先需要定義連接，然后聲明消息隊(duì)列。要接收消息，需要定義一個(gè)Consume，然后從消息隊(duì)列中不斷DeNET environment using RabbitMQ消息，然后處理。

現(xiàn)在發(fā)送端和接收端的代碼都寫好了，運(yùn)行發(fā)送端，發(fā)送消息：

現(xiàn)在，名為hello的消息隊(duì)列中，發(fā)送了一條消息。這條消息存儲(chǔ)到了RabbitMQ的服務(wù)器上了。使用rabbitmqctl 的list_NET environment using RabbitMQs可以查看所有的消息隊(duì)列，以及里面的消息個(gè)數(shù)，可以看到，目前Rabbitmq上只有一個(gè)消息隊(duì)列，里面只有一條消息：

rabbitmqctl list_NET environment using RabbitMQs
Listing NET environment using RabbitMQs ...
hello   1

現(xiàn)在運(yùn)行接收端程序：

可以看到，已經(jīng)接受到了客戶端發(fā)送的Hello World，現(xiàn)在再來(lái)看RabitMQ上的消息隊(duì)列信息：

rabbitmqctl list_NET environment using RabbitMQs
Listing NET environment using RabbitMQs ...
hello   0

可以看到，hello這個(gè)隊(duì)列中的消息隊(duì)列個(gè)數(shù)為0，這表示，當(dāng)接收端，接收到消息之后，RabbitMQ上就把這個(gè)消息刪掉了。

2.2 工作隊(duì)列

前面的例子展示了如何往一個(gè)指定的消息隊(duì)列中發(fā)送和收取消息。現(xiàn)在我們創(chuàng)建一個(gè)工作隊(duì)列（work NET environment using RabbitMQ）來(lái)將一些耗時(shí)的任務(wù)分發(fā)給多個(gè)工作者（workers）：

工作隊(duì)列（work NET environment using RabbitMQs, 又稱任務(wù)隊(duì)列Task Queues）的主要思想是為了避免立即執(zhí)行并等待一些占用大量資源、時(shí)間的操作完成。而是把任務(wù)（Task）當(dāng)作消息發(fā)送到隊(duì)列中，稍后處理。一個(gè)運(yùn)行在后臺(tái)的工作者（worker）進(jìn)程就會(huì)取出任務(wù)然后處理。當(dāng)運(yùn)行多個(gè)工作者（workers）時(shí)，任務(wù)會(huì)在它們之間共享。

這個(gè)在網(wǎng)絡(luò)應(yīng)用中非常有用，它可以在短暫的HTTP請(qǐng)求中處理一些復(fù)雜的任務(wù)。在一些實(shí)時(shí)性要求不太高的地方，我們可以處理完主要操作之后，以消息的方式來(lái)處理其他的不緊要的操作，比如寫日志等等。

準(zhǔn)備

在第一部分，發(fā)送了一個(gè)包含“Hello World!”的字符串消息?，F(xiàn)在發(fā)送一些字符串，把這些字符串當(dāng)作復(fù)雜的任務(wù)。這里使用time.sleep()函數(shù)來(lái)模擬耗時(shí)的任務(wù)。在字符串中加上點(diǎn)號(hào)（.）來(lái)表示任務(wù)的復(fù)雜程度，一個(gè)點(diǎn)（.）將會(huì)耗時(shí)1秒鐘。比如"Hello..."就會(huì)耗時(shí)3秒鐘。

對(duì)之前示例的send.cs做些簡(jiǎn)單的調(diào)整，以便可以發(fā)送隨意的消息。這個(gè)程序會(huì)按照計(jì)劃發(fā)送任務(wù)到我們的工作隊(duì)列中。

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 = <strong>GetMessage(args);</strong>
          <strong>  var properties = channel.CreateBasicProperties();
            properties.DeliveryMode = 2;</strong>

            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!");
}

加粗部分是經(jīng)過(guò)修改過(guò)了的。

接著我們修改接收端，讓他根據(jù)消息中的逗點(diǎn)的個(gè)數(shù)來(lái)Sleep對(duì)應(yīng)的秒數(shù)：

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 NET environment using RabbitMQ = new QueueingBasicConsumer(channel);
            channel.BasicConsume("hello", true, NET environment using RabbitMQ);

            while (true)
            {
                var ea = (BasicDeliverEventArgs)NET environment using RabbitMQ.Queue.DeNET environment using RabbitMQ();

                var body = ea.Body;
                var message = Encoding.UTF8.GetString(body);

              <strong>  int dots = message.Split('.').Length - 1;
                Thread.Sleep(dots * 1000);</strong>
                        
                Console.WriteLine("Received {0}", message);
                Console.WriteLine("Done");
            }
        }
    }
}

輪詢分發(fā)

使用工作隊(duì)列的一個(gè)好處就是它能夠并行的處理隊(duì)列。如果堆積了很多任務(wù)，我們只需要添加更多的工作者（workers）就可以了，擴(kuò)展很簡(jiǎn)單。

現(xiàn)在，我們先啟動(dòng)兩個(gè)接收端，等待接受消息，然后啟動(dòng)一個(gè)發(fā)送端開(kāi)始發(fā)送消息。

在cmd條件下，發(fā)送了5條消息，每條消息后面的逗點(diǎn)表示該消息需要執(zhí)行的時(shí)長(zhǎng)，來(lái)模擬耗時(shí)的操作。

然后可以看到，兩個(gè)接收端依次接收到了發(fā)出的消息：

默認(rèn)，RabbitMQ會(huì)將每個(gè)消息按照順序依次分發(fā)給下一個(gè)消費(fèi)者。所以每個(gè)消費(fèi)者接收到的消息個(gè)數(shù)大致是平均的。 這種消息分發(fā)的方式稱之為輪詢（round-robin）。

2.3 消息響應(yīng)

當(dāng)處理一個(gè)比較耗時(shí)得任務(wù)的時(shí)候，也許想知道消費(fèi)者（NET environment using RabbitMQs）是否運(yùn)行到一半就掛掉。在當(dāng)前的代碼中，當(dāng)RabbitMQ將消息發(fā)送給消費(fèi)者（NET environment using RabbitMQs）之后，馬上就會(huì)將該消息從隊(duì)列中移除。此時(shí)，如果把處理這個(gè)消息的工作者（worker）停掉，正在處理的這條消息就會(huì)丟失。同時(shí)，所有發(fā)送到這個(gè)工作者的還沒(méi)有處理的消息都會(huì)丟失。

我們不想丟失任何任務(wù)消息。如果一個(gè)工作者（worker）掛掉了，我們希望該消息會(huì)重新發(fā)送給其他的工作者（worker）。

為了防止消息丟失，RabbitMQ提供了消息響應(yīng)（acknowledgments）機(jī)制。消費(fèi)者會(huì)通過(guò)一個(gè)ack（響應(yīng)），告訴RabbitMQ已經(jīng)收到并處理了某條消息，然后RabbitMQ才會(huì)釋放并刪除這條消息。

如果消費(fèi)者（NET environment using RabbitMQ）掛掉了，沒(méi)有發(fā)送響應(yīng)，RabbitMQ就會(huì)認(rèn)為消息沒(méi)有被完全處理，然后重新發(fā)送給其他消費(fèi)者（NET environment using RabbitMQ）。這樣，即使工作者（workers）偶爾的掛掉，也不會(huì)丟失消息。

消息是沒(méi)有超時(shí)這個(gè)概念的；當(dāng)工作者與它斷開(kāi)連的時(shí)候，RabbitMQ會(huì)重新發(fā)送消息。這樣在處理一個(gè)耗時(shí)非常長(zhǎng)的消息任務(wù)的時(shí)候就不會(huì)出問(wèn)題了。

消息響應(yīng)默認(rèn)是開(kāi)啟的。在之前的例子中使用了no_ack=True標(biāo)識(shí)把它關(guān)閉。是時(shí)候移除這個(gè)標(biāo)識(shí)了，當(dāng)工作者（worker）完成了任務(wù)，就發(fā)送一個(gè)響應(yīng)。

channel.BasicConsume("hello", <strong>false</strong>, NET environment using RabbitMQ);

while (true)
{
    var ea = (BasicDeliverEventArgs)NET environment using RabbitMQ.Queue.DeNET environment using RabbitMQ();

    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");

    <strong>channel.BasicAck(ea.DeliveryTag, false);</strong>
}

現(xiàn)在,可以保證,即使正在處理消息的工作者被停掉,這些消息也不會(huì)丟失,所有沒(méi)有被應(yīng)答的消息會(huì)被重新發(fā)送給其他工作者.

一個(gè)很常見(jiàn)的錯(cuò)誤就是忘掉了BasicAck這個(gè)方法,這個(gè)錯(cuò)誤很常見(jiàn),但是后果很嚴(yán)重. 當(dāng)客戶端退出時(shí),待處理的消息就會(huì)被重新分發(fā),但是RabitMQ會(huì)消耗越來(lái)越多的內(nèi)存,因?yàn)檫@些沒(méi)有被應(yīng)答的消息不能夠被釋放。調(diào)試這種case，可以使用rabbitmqct打印messages_unacknoledged字段。

rabbitmqctl list_NET environment using RabbitMQs name messages_ready messages_unacknowledged
Listing NET environment using RabbitMQs ...
hello    0       0
...done.

2.4 消息持久化

前面已經(jīng)搞定了即使消費(fèi)者down掉，任務(wù)也不會(huì)丟失，但是，如果RabbitMQ Server停掉了，那么這些消息還是會(huì)丟失。

當(dāng)RabbitMQ Server 關(guān)閉或者崩潰，那么里面存儲(chǔ)的隊(duì)列和消息默認(rèn)是不會(huì)保存下來(lái)的。如果要讓RabbitMQ保存住消息，需要在兩個(gè)地方同時(shí)設(shè)置：需要保證隊(duì)列和消息都是持久化的。

首先，要保證RabbitMQ不會(huì)丟失隊(duì)列，所以要做如下設(shè)置：

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

雖然在語(yǔ)法上是正確的，但是在目前階段是不正確的，因?yàn)槲覀冎耙呀?jīng)定義了一個(gè)非持久化的hello隊(duì)列。RabbitMQ不允許我們使用不同的參數(shù)重新定義一個(gè)已經(jīng)存在的同名隊(duì)列，如果這樣做就會(huì)報(bào)錯(cuò)?，F(xiàn)在，定義另外一個(gè)不同名稱的隊(duì)列：

bool durable = true;
channel.NET environment using RabbitMQDeclare("task_NET environment using RabbitMQ", durable, false, false, null);

NET environment using RabbitMQDeclare 這個(gè)改動(dòng)需要在發(fā)送端和接收端同時(shí)設(shè)置。

現(xiàn)在保證了task_NET environment using RabbitMQ這個(gè)消息隊(duì)列即使在RabbitMQ Server重啟之后，隊(duì)列也不會(huì)丟失。 然后需要保證消息也是持久化的， 這可以通過(guò)設(shè)置IBasicProperties.SetPersistent 為true來(lái)實(shí)現(xiàn)：

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

需要注意的是，將消息設(shè)置為持久化并不能完全保證消息不丟失。雖然他告訴RabbitMQ將消息保存到磁盤上，但是在RabbitMQ接收到消息和將其保存到磁盤上這之間仍然有一個(gè)小的時(shí)間窗口。 RabbitMQ 可能只是將消息保存到了緩存中，并沒(méi)有將其寫入到磁盤上。持久化是不能夠一定保證的，但是對(duì)于一個(gè)簡(jiǎn)單任務(wù)隊(duì)列來(lái)說(shuō)已經(jīng)足夠。如果需要消息隊(duì)列持久化的強(qiáng)保證，可以使用publisher confirms

2.5 公平分發(fā)

你可能會(huì)注意到，消息的分發(fā)可能并沒(méi)有如我們想要的那樣公平分配。比如，對(duì)于兩個(gè)工作者。當(dāng)奇數(shù)個(gè)消息的任務(wù)比較重，但是偶數(shù)個(gè)消息任務(wù)比較輕時(shí)，奇數(shù)個(gè)工作者始終處理忙碌狀態(tài)，而偶數(shù)個(gè)工作者始終處理空閑狀態(tài)。但是RabbitMQ并不知道這些，他仍然會(huì)平均依次的分發(fā)消息。

為了改變這一狀態(tài)，我們可以使用basicQos方法，設(shè)置perfetchCount=1 。這樣就告訴RabbitMQ 不要在同一時(shí)間給一個(gè)工作者發(fā)送多于1個(gè)的消息，或者換句話說(shuō)。在一個(gè)工作者還在處理消息，并且沒(méi)有響應(yīng)消息之前，不要給他分發(fā)新的消息。相反，將這條新的消息發(fā)送給下一個(gè)不那么忙碌的工作者。

channel.BasicQos(0, 1, false); 

2.6 完整實(shí)例

現(xiàn)在將所有這些放在一起：

發(fā)送端代碼如下：

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_NET environment using RabbitMQ", durable, false, false, null);
                    
            string message = GetMessage(args);
            var properties = channel.CreateBasicProperties();
            properties.SetPersistent(true);
                  

            var body = Encoding.UTF8.GetBytes(message);
            channel.BasicPublish("", "task_NET environment using RabbitMQ", 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_NET environment using RabbitMQ", durable, false, false, null);
            channel.BasicQos(0, 1, false);

            var NET environment using RabbitMQ = new QueueingBasicConsumer(channel);
            channel.BasicConsume("task_NET environment using RabbitMQ", false, NET environment using RabbitMQ);

            while (true)
            {
                var ea = (BasicDeliverEventArgs)NET environment using RabbitMQ.Queue.DeNET environment using RabbitMQ();

                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還有一個(gè)管理界面，通過(guò)該界面可以查看RabbitMQ Server 當(dāng)前的狀態(tài)，該界面是以插件形式提供的，并且在安裝RabbitMQ的時(shí)候已經(jīng)自帶了該插件。需要做的是在RabbitMQ控制臺(tái)界面中啟用該插件，命令如下：

rabbitmq-plugins enable rabbitmq_management

現(xiàn)在，在瀏覽器中輸入 http://server-name:15672/ server-name換成機(jī)器地址或者域名，如果是本地的，直接用localhost在輸入之后，彈出登錄界面，使用我們之前創(chuàng)建的用戶登錄。

?.

在該界面上可以看到當(dāng)前RabbitMQServer的所有狀態(tài)。

四 總結(jié)

本文簡(jiǎn)單介紹了消息隊(duì)列的相關(guān)概念，并介紹了RabbitMQ消息代理的基本原理以及在Windows 上如何安裝RabbitMQ和在.NET中如何使用RabbitMQ。消息隊(duì)列在構(gòu)建分布式系統(tǒng)和提高系統(tǒng)的可擴(kuò)展性和響應(yīng)性方面有著很重要的作用，希望本文對(duì)您了解消息隊(duì)列以及如何使用RabbitMQ有所幫助。

五 參考文獻(xiàn)

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