先决条件
本教程假定RabbitMQ已经安装,并运行在localhost
标准端口(5672)。如果你使用不同的主机、端口或证书,则需要调整连接设置。
从哪里获得帮助
如果您在阅读本教程时遇到困难,可以通过邮件列表联系我们。
路由
(使用.NET客户端)
在 绑定是指交换器和队列之间的关联关系。可以简单地理解为:某个队列对来自此交换器的消息感兴趣。 绑定可以采用一个额外的 binding key 的含义取决于交换器类型。像我们前面使用的 在上篇教程中,我们的日志记录系统会把所有消息广播给所有消费者,现在我们想要扩展使其可以根据消息的严重性过滤消息。例如,我们希望将日志消息写入磁盘的脚本仅接收严重错误的消息,而不是在警告或者信息类型的消息上浪费磁盘空间。 之前我们使用的是 现在我们将用 在上面的设置中,我们可以看到 如此设置,发布使用路由键 使用相同的 binding key 绑定多个队列是完全合法的。在我们的例子中,我们可以在 我们将为日志系统使用以上模型,我们会在发送消息时使用 direct 型交换机,而不是 fanout 型。我们会基于日志的严重性作为路由键,这样的话接收端脚本将可以选择它期望接收的严重性。让我们首先聚焦在发送日志方面。 我们将要在我们日志系统使用上面的模型,我们会发送消息到一个 一如既往,我们需要先创建一个交换器: 准备好发送消息: 简单起见,我们假定 接收消息将像上一个教程一样工作,但有一个例外 - 我们将为我们感兴趣的每个严重性创建一个新的绑定。 就像之前教程一样,接收消息这一块都运行得很好,唯独有一处不同 -- 那就是我们 为自己所感兴趣的(日志的)每一种严重性创建一个新的绑定。 马上就可以像前面的教程中那样接收消息了,但是有一个不同点,我们要用自己感兴趣的所有日志严重程度分别创建一个绑定。 马上就可以像前面的教程接收消息了,但有一点不同, 我们需要为我们感兴趣的每种日志严重性建立一个新的绑定。 请像往常一样创建项目(请参阅教程[1])。 如果您只想将 如果您想在屏幕上看到所有日志消息,请打开一个新终端并执行以下操作: 例如,想要发出 EmitLogDirect.cs 和 ReceiveLogsDirect.cs 的完整源代码。 跳转到教程[5],了解如何基于模式监听消息。 本文翻译自RabbitMQ官方教程C#版本。本文介绍如与官方有所出入,请以官方最新内容为准。 水平有限,翻译的不好请见谅,如有翻译错误还请指正。 实验环境:RabbitMQ 3.7.4 、.NET Core 2.1.3、Visual Studio Code 最后更新:2018-08-31 作者:Esofar 出处:https://www.cnblogs.com/esofar/p/rabbitmq-routing.htmlroutingKey
参数,为了避免与BasicPublish
方法中相同参数混淆,我们将其称为binding key
(译者注:这里是指路由键从声明角度的一种别称)。下面即是如何使用 binding key 建立一个绑定:channel.QueueBind(queue: queueName, exchange: "direct_logs", routingKey: "black");
fanout
型交换器,忽略了它的值(译者注:依据fanout
交换器的特性,它会没头脑地把消息广播到所有订阅的队列,所以就算指定routingKey
也不会根据其过滤消息)。Direct型交换器
fanout
交换器,但是它没有给我们足够的灵活性 - 它只能进行无意识的广播。direct
交换器替换,direct
交换器背后的路由算法很简单 - 消息会进入其binding key
恰好与routing key
相匹配的队列。
为了说明这一点,请参考以下设置:direct
型交换器X
与两个队列绑定。第一个队列通过键orange
绑定,第二个队列有两个绑定,一个通过键black
绑定、另外一个通过键green
绑定。orange
的消息到交换器最终会被路由到队列Q1
,路由键为black
或green
的消息会去向队列Q2
,而其他所有的消息会被丢弃。多重绑定
X
和Q1
之间添加一个键为black
的绑定。这种情况下,direct
交换器会像fanout
交换器一样,把消息广播到所有匹配的队列,路由键为black
的消息会被分别传送到队列Q1
和Q2
。发送日志
direct
型交换器,而不是fanout
型交换器。我们会使用日志的严重性作为routing key
。这样的话,接收脚本就可以选择性地接收它期望严重性的消息。首先我们来关注如何发送日志。channel.ExchangeDeclare(exchange: "direct_logs", type: ExchangeType.Direct);
var body = Encoding.UTF8.GetBytes(message);
channel.BasicPublish(exchange: "direct_logs",
routingKey: severity,
basicProperties: null, body: body);
severity
可以是info
、warning
或error
任意一值。订阅
var queueName = channel.QueueDeclare().QueueName;foreach(var severity in args)
{
channel.QueueBind(queue: queueName,
exchange: "direct_logs",
routingKey: severity);
}
组合在一起
EmitLogDirect.cs
类的代码:using System;using System.Linq;using RabbitMQ.Client;using System.Text;class EmitLogDirect{ 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: "direct_logs",
type: "direct"); var severity = (args.Length > 0) ? args[0] : "info"; var message = (args.Length > 1)
? string.Join(" ", args.Skip(1).ToArray())
: "Hello World!"; var body = Encoding.UTF8.GetBytes(message);
channel.BasicPublish(exchange: "direct_logs",
routingKey: severity,
basicProperties: null,
body: body);
Console.WriteLine(" [x] Sent '{0}':'{1}'", severity, message);
}
Console.WriteLine(" Press [enter] to exit.");
Console.ReadLine();
}
}
ReceiveLogsDirect.cs
类的代码:using System;using RabbitMQ.Client;using RabbitMQ.Client.Events;using System.Text;class ReceiveLogsDirect{ 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: "direct_logs",
type: "direct"); var queueName = channel.QueueDeclare().QueueName; if(args.Length < 1)
{
Console.Error.WriteLine("Usage: {0} [info] [warning] [error]",
Environment.GetCommandLineArgs()[0]);
Console.WriteLine(" Press [enter] to exit.");
Console.ReadLine();
Environment.ExitCode = 1; return;
} foreach(var severity in args)
{
channel.QueueBind(queue: queueName,
exchange: "direct_logs",
routingKey: severity);
}
Console.WriteLine(" [*] Waiting for messages."); 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();
}
}
}
warning
和error
(不包括info
)日志消息保存到文件,只需打开控制台并输入:cd ReceiveLogsDirect
dotnet run warning error > logs_from_rabbit.log
cd ReceiveLogsDirect
dotnet run info warning error# => [*] Waiting for logs. To exit press CTRL+C
error
日志消息,只需要输入:cd EmitLogDirect
dotnet run error "Run. Run. Or it will explode."# => [x] Sent 'error':'Run. Run. Or it will explode.'
写在最后
共同學習,寫下你的評論
評論加載中...
作者其他優(yōu)質文章