[译]RabbitMQ教程C#版 - 路由

本文涉及的产品
日志服务 SLS,月写入数据量 50GB 1个月
简介: 先决条件本教程假定RabbitMQ已经安装,并运行在localhost标准端口(5672)。如果你使用不同的主机、端口或证书,则需要调整连接设置。从哪里获得帮助如果您在阅读本教程时遇到困难,可以通过邮件列表 联系我们。

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

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

路由

(使用.NET客户端)

教程[3] 中,我们构建了一个简单的日志系统,可以向多个接收者广播消息。

在本教程中,我们会为日志系统再添加一个特性,使其可以只订阅消息的一个子集。例如,将所有日志消息打印到
控制台,同时只会将严重错误消息写入日志文件(保存到磁盘空间)。

绑定

在前面的例子中,我们创建过绑定。不知道您是否还记得下面的代码:

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

绑定是指交换器和队列之间的关联关系。可以简单地理解为:某个队列对来自此交换器的消息感兴趣。

绑定可以采用额外的routingKey参数,为了避免与BasicPublish方法中相同参数混淆,我们将其称为binding key(这里是指路由键从声明角度的一种别称,绑定键)。下面即是如何使用绑定键 建立一个绑定:

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

绑定键的含义取决于交换器类型。像我们前面使用的fanout交换器,忽略了它的值(依据fanout交换器的特性,它会把消息广播到所有订阅的队列,所以就算指定routingKey也不会根据其过滤消息)。

Direct交换器

在上篇教程中,我们的日志系统会把所有消息广播给所有消费者,现在我们想要扩展使其可以根据消息的严重性过滤消息。例如,我们希望将日志消息写入磁盘的脚本仅接收严重错误的消息,而不是在警告或者信息类型的消息上浪费磁盘空间。

之前我们使用的是fanout交换器,它没有给我们足够的灵活性 - 它只能进行无意识的广播。

现在我们要用direct交换器替换它,direct交换器背后的路由算法很简单 - 消息会进入其binding key恰好与routing key相匹配的队列。
为了说明这一点,请参考以下设置:

direct-exchange

在上面的设置中,我们可以看到direct交换器X与两个队列绑定。第一个队列通过键orange绑定,第二个队列有两个绑定,一个通过键black绑定、另外一个通过键green绑定。

如此设置,发布使用路由键orange的消息到交换器最终会被路由到队列Q1,路由键为blackgreen的消息会去向队列Q2,而其他所有的消息会被丢弃。

多重绑定

direct-exchange-multiple

使用相同的绑定键绑定多个队列是完全合法的。在示例中,我们可以在XQ1之间添加一个键为black的绑定。这种情况下,direct交换器会像fanout交换器一样,把消息广播到所有匹配的队列,路由键为black的消息会被分别发送到队列Q1Q2

发送日志

我们将在日志系统中使用上述消息模型,在发送消息时使用direct交换机来替换fanout交换器。同时我们会把日志的严重性作为路由键,这样的话,接收脚本就可以选择性地接收它期望严重性的消息。首先我们来关注如何发送日志。

同样地,我们需要先创建一个交换器:

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可以是infowarningerror任意一值。

订阅

马上就可以像前面的教程接收消息了,但有一点不同, 我们需要为我们感兴趣的每种日志严重性级别的消息建立一个新的绑定。

var queueName = channel.QueueDeclare().QueueName;

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

组合在一起

image

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();
        }
    }
}

请像往常一样创建项目(请参阅 教程[1])。

如果您想将warningerror(不包括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.'

EmitLogDirect.csReceiveLogsDirect.cs 的完整源代码。

跳转到 教程[5],了解如何基于模式监听消息。

写在最后

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

  • 原文链接:RabbitMQ tutorial - Routing
  • 实验环境:RabbitMQ 3.7.4 、.NET Core 2.1.3、Visual Studio Code
  • 最后更新:2018-08-31
相关实践学习
消息队列RocketMQ版:基础消息收发功能体验
本实验场景介绍消息队列RocketMQ版的基础消息收发功能,涵盖实例创建、Topic、Group资源创建以及消息收发体验等基础功能模块。
消息队列 MNS 入门课程
1、消息队列MNS简介 本节课介绍消息队列的MNS的基础概念 2、消息队列MNS特性 本节课介绍消息队列的MNS的主要特性 3、MNS的最佳实践及场景应用 本节课介绍消息队列的MNS的最佳实践及场景应用案例 4、手把手系列:消息队列MNS实操讲 本节课介绍消息队列的MNS的实际操作演示 5、动手实验:基于MNS,0基础轻松构建 Web Client 本节课带您一起基于MNS,0基础轻松构建 Web Client
目录
相关文章
|
4月前
|
消息中间件 Java RocketMQ
RocketMQ实战教程之RocketMQ安装
这是一篇关于RocketMQ安装的实战教程,主要介绍了在CentOS系统上使用传统安装和Docker两种方式安装RocketMQ。首先,系统需要是64位,并且已经安装了JDK 1.8。传统安装包括下载安装包,解压并启动NameServer和Broker。Docker安装则涉及安装docker和docker-compose,然后通过docker-compose.yaml文件配置并启动服务。教程还提供了启动命令和解决问题的提示。
|
14天前
|
设计模式 C# 开发者
C#设计模式入门实战教程
C#设计模式入门实战教程
|
1月前
|
网络协议 物联网 测试技术
App Inventor 2 MQTT拓展入门(保姆级教程)
本文演示的是App和一个测试客户端进行消息交互的案例,实际应用中,我们的测试客户端可以看着是任意的、支持MQTT协议的硬件,通过订阅及发布消息,联网硬件与我们的App进行双向数据通信,以实现万物互联的智能控制效果。
126 2
|
27天前
|
消息中间件 监控 Ubuntu
RabbitMQ安装配置,超详细版教程
以上步骤为您提供了在Linux环境下安装RabbitMQ的详细过程。安装Erlang作为基础,然后通过添加官方源并安装RabbitMQ本身,最后对服务进行配置并启用Web管理界面。这些步骤操作简单直观,只需要跟随上述指南,即可在短时间内将RabbitMQ服务器运行起来,并进行进一步的配置和管理。不要忘记硬件和网络资源对性能的影响,确保RabbitMQ能够满足您的应用需求。
76 0
|
2月前
|
消息中间件 Kafka
面试题Kafka问题之RabbitMQ的路由配置工作如何解决
面试题Kafka问题之RabbitMQ的路由配置工作如何解决
48 1
|
2月前
|
机器学习/深度学习 算法 搜索推荐
一个开源且全面的C#算法实战教程
一个开源且全面的C#算法实战教程
|
4月前
|
消息中间件 存储 Apache
RocketMQ实战教程之常见概念和模型
Apache RocketMQ 实战教程介绍了其核心概念和模型。消息是基本的数据传输单元,主题是消息的分类容器,支持字节、数字和短划线命名,最长64个字符。消息类型包括普通、顺序、事务和定时/延时消息。消息队列是实际存储和传输消息的容器,是主题的分区。消费者分组是一组行为一致的消费者的逻辑集合,也有命名限制。此外,文档还提到了一些使用约束和建议,如主题和消费者组名的命名规则,消息大小限制,请求超时时间等。RocketMQ 提供了多种消息模型,包括发布/订阅模型,有助于理解和优化消息处理。
|
3月前
|
消息中间件 Java RocketMQ
教程:Spring Boot整合RocketMQ的配置与优化
教程:Spring Boot整合RocketMQ的配置与优化
|
4月前
|
消息中间件 存储 Java
RocketMQ实战教程之NameServer与BrokerServer
这是一个关于RocketMQ实战教程的概要,主要讨论NameServer和BrokerServer的角色。NameServer负责管理所有BrokerServer,而BrokerServer存储和传输消息。生产者和消费者通过NameServer找到合适的Broker进行交互,不需要直接知道Broker的具体信息。工作流程包括生产者向NameServer查询后发送消息到Broker,以及消费者同样通过NameServer获取消息进行消费。这种设计类似于服务注册中心的概念,便于系统扩展和集群管理。
|
3月前
|
消息中间件 Java Spring
最新spingboot整合rabbitmq详细教程
最新spingboot整合rabbitmq详细教程