RabbitMQ 教程 - 主题
主题
（使用 .NET 客户端）
在上一教程中，我们改进了我们的日志系统。我们没有只使用只能进行虚拟广播的 fanout 交换机，而是使用了 direct 交换机，并获得了选择性接收日志的可能性。
尽管使用 direct 交换机改进了我们的系统，但它仍然存在局限性 - 它无法基于多个条件进行路由。
在我们的日志系统中，我们可能不仅希望订阅基于严重性的日志，还希望订阅基于发出日志的源的日志。您可能从syslog unix 工具中了解过此概念，该工具根据严重性（info/warn/crit...）和设备（auth/cron/kern...）对日志进行路由。
这将为我们提供很大的灵活性 - 我们可能希望监听来自 'cron' 的所有严重错误，以及来自 'kern' 的所有日志。
为了在我们的日志系统中实现这一点，我们需要了解更复杂的 topic 交换机。
主题交换机
发送到 topic 交换机的消息不能具有任意的 routing_key - 它必须是单词列表，用点分隔。单词可以是任何内容，但通常它们指定与消息相关的一些功能。一些有效的路由键示例：stock.usd.nyse、nyse.vmw、quick.orange.rabbit。路由键中的单词数量可以任意多，最多 255 个字节。
绑定键也必须采用相同的形式。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.new.rabbit 会发生什么？好吧，这些消息将不匹配任何绑定，并将丢失。
另一方面，lazy.orange.new.rabbit 即使有四个单词，也将匹配最后一个绑定，并将传递到第二个队列。
主题交换机
主题交换机功能强大，可以像其他交换机一样工作。
当队列使用
#（井号）绑定键绑定时 - 它将接收所有消息，而不管路由键是什么 - 就像fanout交换机一样。当在绑定中不使用特殊字符
*（星号）和#（井号）时，主题交换机将表现得像direct交换机一样。
综合起来
我们将在日志系统中使用 topic 交换机。我们将从一个工作假设开始，即日志的路由键将有两个单词：<facility>.<severity>。
代码与上一教程中的代码几乎相同。
EmitLogTopic.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: "topic_logs", type: ExchangeType.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 '{routingKey}':'{message}'");
ReceiveLogsTopic.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: "topic_logs", type: ExchangeType.Topic);
// declare a server-named queue
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.ToArray();
var message = Encoding.UTF8.GetString(body);
var routingKey = ea.RoutingKey;
Console.WriteLine($" [x] Received '{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中将往返消息作为远程过程调用。