[译]RabbitMQ教程C#版 - 主题

2021-12-04 03:28:13

先决条件
本教程假定RabbitMQ已经安装，并运行在localhost标准端口（5672）。如果你使用不同的主机、端口或证书，则需要调整连接设置。

从哪里获得帮助
如果您在阅读本教程时遇到困难，可以通过邮件列表联系我们。

主题

（使用.NET客户端）

在教程[4] 中，我们改进了我们日志系统。我们用direct交换器替换了只能呆滞广播消息的fanout交换器，从而可以有选择性的接收日志。

虽然使用direct交换器改进了我们的系统，但它仍然有局限性 - 不能基于多个标准进行路由。

在我们的日志系统中，我们可能不仅要根据日志的严重性订阅日志，可能还要根据日志分发源来订阅日志。或许您可能从 unix syslog 工具中了解过这种概念，syslog 工具在路由日志的时候是可以既基于严重性（info/warn/crit...）又基于设备（auth/cron/kern...）的。

这种机制会给我们带来极大的灵活性 - 我们可以仅监听来自cron的关键错误日志，与此同时，监听来自kern的所有日志。

要在我们的日志系统中实现这一特性，我们需要学习更复杂的topic交换器。

Topic交换器

发送到topic交换器的消息不能随意指定routing key，它必须是一个由点分割的单词列表，这些单词可以是任意内容，但通常会在其中指定一些与消息相关的特性。请看一些合法的路由键示例：stock.usd.nyse，nyse.vmw，quick.orange.rabbit，路由键可以包含任意数量的单词，但不能超过255个字节的上限。

binding key也必须是相同的形式，topic交换器的背后逻辑与direct交换器类似 - 使用指定路由键发送的消息会被分发到与其绑定键匹配的所有队列中。不过对于绑定键来说，有两个重要的特殊情况需要注意：

*（星号）可以代替一个单词。
#（哈希）可以代替零个或多个单词。

下图示例是对上述内容最简单的解释：

在这个示例中，我们打算发送的消息全是用来描述动物的，这些消息会使用由三个单词（两个点）组成的路由键来发送。在路由键中，第一个单词用来描述行动速度、第二个是颜色、第三个是物种，即：<speed>.<colour>.<species>。

我们创建了三个绑定：Q1绑定了键.orange.，Q2绑定了键*.*.rabbit和lazy.#。

这些绑定可以被概括为：

Q1对所有橙色的动物感兴趣。
Q2对兔子以及所有行动缓慢的动物感兴趣。

路由键为quick.orange.rabbit的消息会被发送到这两个队列，消息lazy.orange.elephant也会被发送到这两个队列。另外，quick.orange.fox只会进入第一个队列，lazy.brown.fox只会进入第二个队列。lazy.pink.rabbit只会被发送到第二个队列一次，尽管它匹配了两个绑定（避免了消息重复）。quick.brown.fox没有匹配的绑定，因此它将会被丢弃。

如果我们打破约定，发送使用一个或四个单词（例如：orange和quick.orange.male.rabbit）作路由键的消息会发生什么？答案是，这些消息因为没有匹配到任何绑定，将被丢弃。

但是，另外，例如路由键为lazy.orange.male.rabbit的消息，尽管它有四个单词，也会匹配最后一个绑定，并将被发送到第二个队列。

Topics 交换器

topic交换器的功能是很强大的，它可以表现出一些其他交换器的行为。

当一个队列与键＃（哈希）绑定时，它会忽略路由键，接收所有消息，这就像fanout交换器一样。

当特殊字符*（星号）和＃（哈希）未在绑定中使用时，topic交换器的行为就像direct交换器一样。

组合在一起

我们将要在我们的日志系统中使用topic交换器，首先假设日志的路由键有两个单词组成：<facility>.<severity>。

代码与上一篇教程中的代码几乎相同。

EmitLogTopic.cs的代码：

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

class EmitLogTopic
{
    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: "topic_logs",
                                    type: "topic");

            var routingKey = (args.Length > 0) ? args[0] : "anonymous.info";
            
            var message = (args.Length > 1)
                          ? string.Join(" ", args.Skip(1).ToArray())
                          : "Hello World!";
            var body = Encoding.UTF8.GetBytes(message);
            
            channel.BasicPublish(exchange: "topic_logs",
                                 routingKey: routingKey,
                                 basicProperties: null,
                                 body: body);
                                 
            Console.WriteLine(" [x] Sent '{0}':'{1}'", routingKey, message);
        }
    }
}

ReceiveLogsTopic.cs的代码：

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

class ReceiveLogsTopic
{
    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: "topic_logs", type: "topic");
            var queueName = channel.QueueDeclare().QueueName;

            if(args.Length < 1)
            {
                Console.Error.WriteLine("Usage: {0} [binding_key...]",
                                        Environment.GetCommandLineArgs()[0]);
                Console.WriteLine(" Press [enter] to exit.");
                Console.ReadLine();
                Environment.ExitCode = 1;
                return;
            }

            foreach(var bindingKey in args)
            {
                channel.QueueBind(queue: queueName,
                                  exchange: "topic_logs",
                                  routingKey: bindingKey);
            }

            Console.WriteLine(" [*] Waiting for messages. To exit press CTRL+C");

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

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

请运行以下示例：

要接收所有日志：

cd ReceiveLogsTopic
dotnet run "#"

要接收来自设备kern的所有日志：

cd ReceiveLogsTopic
dotnet run "kern.*"

或者，如果您只想监听级别为critical的日志：

cd ReceiveLogsTopic
dotnet run "*.critical"

您可以创建多个绑定：

cd ReceiveLogsTopic
dotnet run "kern.*" "*.critical"

使用路由键kern.critical发出日志：

cd EmitLogTopic
dotnet run "kern.critical" "A critical kernel error"

希望运行这些程序能让您玩得开心。要注意的是，这些代码没有针对路由键和绑定键做任何预设，您可以尝试使用两个以上的路由键参数。

（ EmitLogTopic.cs 和 ReceiveLogsTopic.cs 的完整源码）

接下来，在教程[6] 中将了解如何将往返消息作为远程过程调用。

写在最后

本文翻译自RabbitMQ官方教程C#版本。如本文介绍内容与官方有所出入，请以官方最新内容为准。水平有限，翻译的不好请见谅，如有翻译错误还请指正。

原文链接：RabbitMQ tutorial - Topics
实验环境：RabbitMQ 3.7.4 、.NET Core 2.1.3、Visual Studio Code
最后更新：2018-09-06

码农公寓

主题

Topic交换器

组合在一起

写在最后

相关文章