《RabbitMQ Tutorial》譯文 第 3 章 發佈和訂閱

原文來自 RabbitMQ 英文官網的教程（3.Publish and Subscribe），其示例代碼採用了 .NET C# 語言。

In the previous tutorial we created a work queue. The assumption behind a work queue is that each task is delivered to exactly one worker. In this part we'll do something completely different -- we'll deliver a message to multiple consumers. This pattern is known as "publish/subscribe".

在以前的教程中咱們建立了一個工做隊列，其背後的設想即是每個任務剛好地遞送給一個工做單元。在本教程中，咱們的作法將徹底不一樣 -- 即遞送消息給多個消費者，這個模式被稱做「發佈/訂閱」。

To illustrate the pattern, we're going to build a simple logging system. It will consist of two programs -- the first will emit log messages and the second will receive and print them.

爲了說明這個模式，咱們將要構建一個簡單日誌系統。它將由兩個程序組成 -- 第一個將會發送日誌消息，而第二個將會接收並打印它們。

In our logging system every running copy of the receiver program will get the messages. That way we'll be able to run one receiver and direct the logs to disk; and at the same time we'll be able to run another receiver and see the logs on the screen.

在咱們的日誌系統中，每個正在運行的接收程序副本都將得到消息。如此，咱們將運行一個接收者來將日誌寫入磁盤，與此同時再運行另外一個接收者，這樣能夠在屏幕上查看日誌。

Essentially, published log messages are going to be broadcast to all the receivers.

本質上來說，已發佈的日誌消息將會廣播給全部的接收者。

Exchanges

交換機

In previous parts of the tutorial we sent and received messages to and from a queue. Now it's time to introduce the full messaging model in Rabbit.

在本教程的以前部分，咱們從隊列中發送和接收消息，如今是時候來介紹 Rabbit 中完整的消息模型了。

Let's quickly go over what we covered in the previous tutorials:

• A producer is a user application that sends messages.
• A queue is a buffer that stores messages.
• A consumer is a user application that receives messages.

讓咱們快速溫習一下以前教程中所包括的內容：

• 生產者，用以發送消息的用戶程序。
• 隊列，即存儲消息的緩衝區。
• 消費者，用以接收消息的用戶程序。

The core idea in the messaging model in RabbitMQ is that the producer never sends any messages directly to a queue. Actually, quite often the producer doesn't even know if a message will be delivered to any queue at all.

在 RabbitMQ 中，消息模型的核心理念是生產者永遠不會把任何消息直接發送到隊列。事實上，一般生產者甚至不知道消息是否會被遞送到任何隊列。

Instead, the producer can only send messages to an exchange. An exchange is a very simple thing. On one side it receives messages from producers and the other side it pushes them to queues. The exchange must know exactly what to do with a message it receives. Should it be appended to a particular queue? Should it be appended to many queues? Or should it get discarded. The rules for that are defined by the exchange type.

取而代之的是，生產者只能發送消息給一個交換機。交換機很簡單，一方面它從生產者接收消息，另外一方面它又把消息推送到隊列。可是交換機必須明確知道對它所接收到的消息該作什麼。（好比）須要將消息追加到指定隊列？仍是追加到多個隊列？仍是要丟棄？這些規則都是在交換機類型中定義。

There are a few exchange types available: direct, topic, headers and fanout. We'll focus on the last one -- the fanout. Let's create an exchange of this type, and call it logs:

目前有若干交換機類型可用：direct、topic、headers 以及 fanout。咱們將以最後一個爲重點 -- fanout，讓咱們建立一個該類型的交換機，並將其叫做「logs」：

channel.ExchangeDeclare("logs", "fanout");

The fanout exchange is very simple. As you can probably guess from the name, it just broadcasts all the messages it receives to all the queues it knows. And that's exactly what we need for our logger.

fanout 型交換機很是簡單，從它的名稱你可能猜出，它會廣播全部已接收到的消息給全部已知的隊列，這正好是咱們的日誌系統所須要的。

Listing exchanges

列舉出交換機

To list the exchanges on the server you can run the ever useful rabbitmqctl:

爲了列舉出服務端的交換機，你能夠運行此前很是有用的 rabbitmqctl 命令：

sudo rabbitmqctl list_exchanges

In this list there will be some amq.* exchanges and the default (unnamed) exchange. These are created by default, but it is unlikely you'll need to use them at the moment.

在這個列表中有一些 amq.* 和默認的（未命名）交換機，這些都是默認建立的，但此時你不太可能須要用到它們。

The default exchange

默認的交換機

In previous parts of the tutorial we knew nothing about exchanges, but still were able to send messages to queues. That was possible because we were using a default exchange, which we identify by the empty string ("").

在教程的以前部分咱們對交換機還一無所知，但這並不妨礙咱們可以發送消息到隊列中，這之因此成爲可能，是由於咱們使用了基於空字符串來標識的默認交換機。

Recall how we published a message before:

回憶一下之前咱們是如何發佈消息的：

var message = GetMessage(args);
var body = Encoding.UTF8.GetBytes(message);
channel.BasicPublish(exchange: "",
                    routingKey: "hello",
                    basicProperties: null,
                    body: body);

The first parameter is the the name of the exchange. The empty string denotes the default or nameless exchange: messages are routed to the queue with the name specified by routingKey, if it exists.

第一個參數就是交換機的名字。空字符串表示默認的或者匿名的交換機：根據明確的路由鍵（routingKey）將消息路由到已存在的隊列。

Now, we can publish to our named exchange instead:

如今，咱們用具名（自定義命名）的交換機來進行發佈：

var message = GetMessage(args);
var body = Encoding.UTF8.GetBytes(message);
channel.BasicPublish(exchange: "logs",
                     routingKey: "",
                     basicProperties: null,
                     body: body);

Temporary queues

臨時隊列

As you may remember previously we were using queues which had a specified name (remember hello and task_queue?). Being able to name a queue was crucial for us -- we needed to point the workers to the same queue. Giving a queue a name is important when you want to share the queue between producers and consumers.

你可能還記得以前咱們使用過指定名稱的隊列（記得好像是 hello 和 task_queue？）。可以爲隊列命名對咱們來講是相當重要的 -- 咱們須要將工做單元指向相同的隊列。一樣，當你想在生產者和消費者之間共享隊列時，爲隊列賦予一個名字也是很是重要的。

But that's not the case for our logger. We want to hear about all log messages, not just a subset of them. We're also interested only in currently flowing messages not in the old ones. To solve that we need two things.

可是，以上並非咱們日誌系統的案例。咱們想要監聽全部的日誌消息，而不只僅是它們的一部分。咱們只對當前正在流動的消息感興趣，而不是舊的消息，爲解決這個問題我須要作好兩件事。

Firstly, whenever we connect to Rabbit we need a fresh, empty queue. To do this we could create a queue with a random name, or, even better - let the server choose a random queue name for us.

首先，不管咱們什麼時候鏈接到 Rabbit，都須要一個嶄新的、空的隊列，爲作到這一點咱們可使用隨機名稱來建立一個隊列，固然更好的作法是，讓服務端爲咱們選擇一個隨機名稱。

Secondly, once we disconnect the consumer the queue should be automatically deleted.

其次，一旦咱們與消費者斷開鏈接，相關的隊列也應當能自動刪除。

In the .NET client, when we supply no parameters to queueDeclare() we create a non-durable, exclusive, autodelete queue with a generated name:

在 .NET 客戶端中，當咱們調用 queueDeclare 方法而並未提供任何參數時，實際上就是建立了一個非持久化、獨享，且自動刪除的具名隊列。

var queueName = channel.QueueDeclare().QueueName;

At that point queueName contains a random queue name. For example it may look like amq.gen-JzTY20BRgKO-HjmUJj0wLg.

在此處 queueName 包含的是一個隨機隊列名稱，好比它看起來可能像 amq.gen-JzTY20BRgKO-HjmUJj0wLg。

Bindings

綁定

We've already created a fanout exchange and a queue. Now we need to tell the exchange to send messages to our queue. That relationship between exchange and a queue is called a binding.

咱們已經建立了一個 fanout 型交換機和隊列，如今咱們須要告訴交換機把消息發送到隊列，那麼，交換機和隊列之間的關係就被稱做綁定。

channel.QueueBind(queue: queueName,
                  exchange: "logs",
                  routingKey: "");

From now on the logs exchange will append messages to our queue.

從如今開始，日誌交換機將會把消息追加到隊列中。

Listing bindings

列舉綁定

You can list existing bindings using, you guessed it:

你能夠列舉出現有的綁定，（所使用的命令）你該能夠猜到：

rabbitmqctl list_bindings

Putting it all together

融合一塊兒

The producer program, which emits log messages, doesn't look much different from the previous tutorial. The most important change is that we now want to publish messages to our logs exchange instead of the nameless one. We need to supply a routingKey when sending, but its value is ignored for fanout exchanges. Here goes the code for EmitLog.cs file:

發出日誌消息的生產者程序，與以前教程看起來並沒有太大不一樣。如今，最重要的變化莫過於咱們想把消息發佈到名爲 logs 的交換器中，而非匿名。在發送消息時咱們須要提供一個路由鍵（routingKey），只不過它的值在 fanout 型交換機中被忽略了，針對 EmitLog.cs 文件的代碼以下：

using System;
using RabbitMQ.Client;
using System.Text;

class EmitLog
{
    public static void Main(string[] args)
    {
        var factory = new ConnectionFactory() { HostName = "localhost" };
        using(var connection = factory.CreateConnection())
        using(var channel = connection.CreateModel())
        {
            channel.ExchangeDeclare(exchange: "logs", type: "fanout");

            var message = GetMessage(args);
            var body = Encoding.UTF8.GetBytes(message);
            channel.BasicPublish(exchange: "logs",
                                 routingKey: "",
                                 basicProperties: null,
                                 body: body);
            Console.WriteLine(" [x] Sent {0}", message);
        }

        Console.WriteLine(" Press [enter] to exit.");
        Console.ReadLine();
    }

    private static string GetMessage(string[] args)
    {
        return ((args.Length > 0)
               ? string.Join(" ", args)
               : "info: Hello World!");
    }
}

(EmitLog.cs source)

As you see, after establishing the connection we declared the exchange. This step is necessary as publishing to a non-existing exchange is forbidden.

如同你所見，在創建好鏈接以後咱們聲明瞭交換機。這一步很是有必要，由於禁止向一個不存在的交換機發布消息。

The messages will be lost if no queue is bound to the exchange yet, but that's okay for us; if no consumer is listening yet we can safely discard the message.

若是尚沒有任何隊列綁定到交換機，消息將會丟失，但這對咱們來講並無什麼問題；若是沒有任何消費者正在監聽，咱們能夠將消息安全地刪除。

The code for ReceiveLogs.cs:

using System;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using System.Text;

class ReceiveLogs
{
    public static void Main()
    {
        var factory = new ConnectionFactory() { HostName = "localhost" };
        using(var connection = factory.CreateConnection())
        using(var channel = connection.CreateModel())
        {
            channel.ExchangeDeclare(exchange: "logs", type: "fanout");

            var queueName = channel.QueueDeclare().QueueName;
            channel.QueueBind(queue: queueName,
                              exchange: "logs",
                              routingKey: "");

            Console.WriteLine(" [*] Waiting for logs.");

            var consumer = new EventingBasicConsumer(channel);
            consumer.Received += (model, ea) =>
            {
                var body = ea.Body;
                var message = Encoding.UTF8.GetString(body);
                Console.WriteLine(" [x] {0}", message);
            };
            channel.BasicConsume(queue: queueName,
                                 autoAck: true,
                                 consumer: consumer);

            Console.WriteLine(" Press [enter] to exit.");
            Console.ReadLine();
        }
    }
}

(ReceiveLogs.cs source)

Follow the setup instructions from tutorial one to generate the EmitLogs and ReceiveLogs projects.

從教程的第一章開始，跟隨安裝說明來生成 EmitLogs 和 ReceiveLogs 工程。

If you want to save logs to a file, just open a console and type:

若是你想保存日誌到一個文件，只需打開控制檯並輸入：

cd ReceiveLogs
dotnet run > logs_from_rabbit.log

If you wish to see the logs on your screen, spawn a new terminal and run:

若是你想在屏幕上查看日誌，重開一個新的終端並運行：

cd ReceiveLogs
dotnet run

And of course, to emit logs type:

固然，產生日誌只需輸入：

cd EmitLog
dotnet run

Using rabbitmqctl list_bindings you can verify that the code actually creates bindings and queues as we want. With two ReceiveLogs.cs programs running you should see something like:

使用 rabbitmqctl list_bindings 命令你能夠覈實代碼的確已經建立了咱們指望的綁定和隊列，伴隨着 ReceiveLogs.cs 程序的運行你應該能夠看到相似以下內容：

sudo rabbitmqctl list_bindings
# => Listing bindings ...
# => logs    exchange        amq.gen-JzTY20BRgKO-HjmUJj0wLg  queue           []
# => logs    exchange        amq.gen-vso0PVvyiRIL2WoV3i48Yg  queue           []
# => ...done.

The interpretation of the result is straightforward: data from exchange logs goes to two queues with server-assigned names. And that's exactly what we intended.

對結果的解釋就很是簡潔明瞭：來自 logs 交換機的數據將去往兩個由服務端分配名稱的隊列，而這剛好是咱們所指望的。