（三）「消息队列」之 RabbitMQ 发布/订阅（使用 .NET 客户端）_netmq 发布订阅

0、引言

先决条件

本教程假设 RabbitMQ 已安装并且正在 本地主机 的标准端口（5672）上运行。如果您使用了不同的主机、端口或凭证，则要求调整连接设置。

获取帮助

如果您在阅读本教程时遇到问题，可以通过邮件列表或者 RabbitMQ 社区 Slack 与 RabbitMQ 官方取得联系。

在上一篇教程中我们创建了一个工作队列。工作队列背后的设想是，每个任务只交付给一个工作者（worker）。在本篇当中，我们将做一些完全不同的事情 —— 我们将向多个消费者传递一条消息。这种广为人知的模式被称为 “发布/订阅”。

为了说明这个模式，我们将构建一个简单的日志系统。它将由两个程序组成 —— 第一个程序将发出日志消息，第二个程序将接收并打印它们。

在我们的日志系统中，每一个正在运行的接收程序（receiver program, 后面简称为 “接收者”）副本都会得到消息。通过这样的方式我们得以运行一个接收者（receiver）并将日志导入到磁盘中；与此同时我们还可以运行另一个接收者并在屏幕上看到日志。

从本质上讲，发布的日志消息将会被广播给所有的接收者。

原文链接：https://www.rabbitmq.com/tutorials/tutorial-three-dotnet.html

1、交换机

在之前的教程中我们 向(从) 一个队列 发送(接收) 消息。现在，是时候介绍 Rabbit 中完整的消息模型了。

让我们快速回顾下我们在之前的教程中所涵盖的内容：

• 一个 生产者 Producer 是一个发送消息的用户程序。
• 一个 队列 queue 是一个存储消息的缓存器（buffer）。
• 一个 消费者 Consumer 是一个接收消息的用户程序。

RabbitMQ 中消息模型的核心思想是：生产者从不直接向一个队列发送任何消息。实际上，生产者通常甚至不知道消息是否会被传递到任何队列。

相反，生产者只能将消息发送给一个 交换机 exchange。交换机是个非常简单的东西：一方面它从生产者方接收消息；另一方面它将消息推入队列。交换机必须明确地知道要如何处理其收到的消息：应该将消息附加到某个确切的队列中吗？还是应该将消息附加到许多队列中？或者应该丢弃消息？这些规则是由 exchange 类型定义的。

这里有一些可用的 exchange 类型：direct, topic, headers 以及 fanout。我们将关注最后一个 —— fanout。让我们创建一个该类型的交换机并将其命名为 logs：

channel.ExchangeDeclare("logs", ExchangeType.Fanout);
1

扇出（fanout）交换机十分简单。您也许能够通过名字猜到，它只是将其收到的所有消息广播给其所知的所有队列。这正是我们的日志发布者（logger）所需的。

罗列交换机

您可以运行十分有用的 rabbitmqctl 命令以罗列服务器上的交换机：

sudo rabbitmqctl list_exchanges
1

在列表中会有一些 amq.* 交换机以及默认（未命名）交换机。它们是默认创建的，但此刻您不大可能用得上它们。

默认交换机

在之前的教程中我们对交换机一无所知，但仍然能够向队列发送消息。这可能是因为那会儿我们通过空字符串（""）标识使用了默认交换机。

回忆一下之前我们是如何发布消息的：

    var message = GetMessage(args);
    var body = Encoding.UTF8.GetBytes(message);
    channel.BasicPublish(exchange: string.Empty,
                         routingKey: "hello",
                         basicProperties: null,
                         body: body);
1
2
3
4
5
6

第一个参数即交换机名称，空字符串表示默认或者匿名的交换机：如果存在由 routingKey 指定的队列，则消息将被路由到具有该名称的队列。

现在，我们可以将消息发布到指定的交换机：

var message = GetMessage(args);
var body = Encoding.UTF8.GetBytes(message);
channel.BasicPublish(exchange: "logs",
                     routingKey: string.Empty,
                     basicProperties: null,
                     body: body);
1
2
3
4
5
6

2、临时队列

您也许还记得之前我们使用的是具备特定名称的队列（还记得 hello 和 task_queue 吗？）能够命名队列对我们来说至关重要 —— 我们需要将工作者（worker）指向同一个队列。当您希望在生产者和消费者之间共享队列时，为队列指定名称是十分重要的。

但对于我们的日志发布者（logger）来说，这都不是问题。我们想要了解所有的日志消息，而不仅仅是其中的一个子集。我们也只对当前流动的消息感兴趣，而不是之前的消息。要解决这个问题，我们需要做两件事情：

首先，无论我们什么时候连接到 RabbitMQ，我们都需要一个全新的、空的队列。为了做到这一点，我们可以创建一个随机名称的队列；或者，更好的办法是 —— 让消息队列服务器为我们选择一个随机的队列名称。

其次，一旦我们与消费者失去连接，队列应当自动被删除。

在 .NET 客户端中，当我们没有为 QueueDeclare() 方法提供参数时，我们会创建一个非持久存储的、独占的（exclusive）、自动删除的队列，它具备一个随机生成的名称：

var queueName = channel.QueueDeclare().QueueName;
1

您可以在队列指南中了解更多有关 exclusive 标志以及其他队列属性的信息。

此时，queueName 包含了一个随机队列名称。比如它有可能看起来像 amq.gen-JzTY20BRgKO-HjmUJj0wLg。

3、绑定

我们已经创建了一个扇出（fanout）交换机和一个队列。现在我们需要告诉交换机将消息发送给我们的队列。交换机和队列之间的关系被称之为 绑定。

channel.QueueBind(queue: queueName,
                  exchange: "logs",
                  routingKey: string.Empty);
1
2
3

从现在开始 logs 交换机会将消息附加到我们的队列当中。

罗列绑定

您可以使用如下命令罗列已有绑定：

rabbitmqctl list_bindings
1

4、将所有的东西放到一起

发送日志消息的生产者程序与上一篇教程不会有太大不同。最大的改变是现在我们想要发布消息到我们的 logs 交换机而不是匿名交换机。我们需要在发送消息时提供一个 routingKey，但对于 fanout 交换机来说，它的值会被忽略。如下是 EmitLog.cs 文件的代码：

using System.Text;
using RabbitMQ.Client;

var factory = new ConnectionFactory { HostName = "localhost" };
using var connection = factory.CreateConnection();
using var channel = connection.CreateModel();

channel.ExchangeDeclare(exchange: "logs", type: ExchangeType.Fanout);

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

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

static string GetMessage(string[] args)
{
    return ((args.Length > 0) ? string.Join(" ", args) : "info: Hello World!");
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

（EmitLog.cs 源码）

如您所见，在建立连接之后我们声明了交换机。这一步是必要的，因为禁止向不存在的交换机发布消息。

如果还没有队列绑定到交换机，那么消息将会丢失，不过对我们来说那都还好；如果还没有消费者在监听那我们可以安全地丢弃该消息。

ReceiveLogs.cs 代码：

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

var factory = new ConnectionFactory { HostName = "localhost" };
using var connection = factory.CreateConnection();
using var channel = connection.CreateModel();

channel.ExchangeDeclare(exchange: "logs", type: ExchangeType.Fanout);

// declare a server-named queue
var queueName = channel.QueueDeclare().QueueName;
channel.QueueBind(queue: queueName,
                  exchange: "logs",
                  routingKey: string.Empty);

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

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

Console.WriteLine(" Press [enter] to exit.");
Console.ReadLine();
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

（ReceiveLogs.cs 源码）

按照教程一的配置说明生成 EmitLogs 和 ReceiveLogs 项目。

如果您想要将日志保存到文件中，只需要打开一个控制台并输入：

cd ReceiveLogs
dotnet run > logs_from_rabbit.log
1
2

如果您想要在屏幕上显示日志，打开一个新的终端并运行：

cd ReceiveLogs
dotnet run
1
2

啊当然，要发送日志，可以输入：

cd EmitLog
dotnet run
1
2

运行效果：

使用 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.
1
2
3
4
5

对结果的解释很简单：来自 logs 交换机的数据进入两个具有服务器分配名称的队列。这正是我们想要的。

要了解如何侦听消息子集，让我们移步至教程四。

5、生产[非]适用性免责声明

请记住，本教程和其他教程都是教程。他们一次展示一个新概念，可能会有意地过度简化一些东西，而忽略其他东西。例如，为了简洁起见，连接管理、错误处理、连接恢复、并发性和指标收集等主题在很大程度上被省略了。这种简化的代码不应该被认为可以用于生产。

在发布您的应用之前，请先查看其他文档。我们特别推荐以下指南：发布者确认和消费者确认，生产清单和监控。