转载RabbitMQ入门(4)--路由

本文涉及的产品
日志服务 SLS,月写入数据量 50GB 1个月
简介: 路由 (使用Java客户端) 在先前的指南中,我们建立了一个简单的日志系统。我们可以将我们的日志信息广播到多个接收者。 在这部分的指南中,我们将要往其中添加一个功能-让仅仅订阅一个消息的子集成为可能。

路由

(使用Java客户端)

在先前的指南中,我们建立了一个简单的日志系统。我们可以将我们的日志信息广播到多个接收者。 在这部分的指南中,我们将要往其中添加一个功能-让仅仅订阅一个消息的子集成为可能。例如,我们可以直接将关键的错误信息指向到日志文件(保存在爱硬盘空间),同时依旧能打印所有日志信息到平台上。

绑定

在之前的例子里我们已经创建绑定。你可以回顾下代码:

channel.queueBind(queueName, EXCHANGE_NAME, "");

A binding is a relationship between an exchange and a queue. This can be simply read as: the queue is interested in messages from this exchange. 一个绑定是一个交换所和一个队列之间的关系。这个很容易理解为:这个队列是对这 交换机的消息感兴趣。

绑定可以带上额外的路由关键字参数。为了消除对basic_publish参数的迷惑,我们将会将它称之为绑定关键字。以下是我们如何通过一个关键字创建一个绑定:

channel.queueBind(queueName, EXCHANGE_NAME, "black");

这绑定关键字的意义取决于 交换机类型。这fanout 交换机,我们之前使用的那个,仅仅忽略它的值。

直接交换

我们当前的日志系统将所有消息广播到所有消费者。我们想扩展它,让其允许依据其严格的规则过滤消息。例如我们可能想让一个往硬盘中写日志消息的程序仅仅接收关键的错误,而不是将硬盘空间浪费在警告和信息的日志消息上。 我们使用fanout类型的交换机,那个不会给我们太多的灵活性-它仅仅能胜任没头脑的广播。

我们可以使用direct类型的交换机来替代。一个direct交换机背后的路由算法是简单的-一个消息将会进入那些队列的绑定关键字与消息中路由关键字匹配的队列中。

为了说明那个,考虑接下来结构: direct-exchange.png 在这个结构里,我们看见了这direct类型的交换机绑定了两个队列。第一个队列装有orange绑定关键字,这第二个有两个绑定,一个是black绑定关键字并且另一个是green关键之。 在这个结构里,发送到交换机里的消息,其中消息中带路由关键字为orange将要路由到队列Q1上,消息中带路由关键字为blackgreen将路由到队列Q2上。所有其他类型的消息会被丢弃。

多种绑定

direct-exchange-multiple.png 将一个绑定关键字绑定到货个队列上是十分合法的。在我们例子中使用绑定关键字blackXQ1绑定在一起。既然那样,这direct类型的交换机与fanout类型相似,同样会广播这消息到所有符合的队列中。一个路由关键位balck的关键字将会被传递到Q1Q2

发出日志

我们将会为我们的日志系统使用这个模型。使用direct类型的交换机来代替fanout类型,发送消息。由于这路由关键字我们可以严格的记录。接收程序通过这种方式可以严格接收它想接收的。让我们首先关注发布日志。 总之,我们首先需要创建个交换机。

channel.exchangeDeclare(EXCHANGE_NAME, "direct");

我们准备发送一个消息:

channel.basicPublish(EXCHANGE_NAME, severity, null, message.getBytes());

为了简化这个事情,我们保证这severityinof,warning,error中的一个。

订阅

接收消息如先前那样工作,有一个例外,我们会把每一个我们感兴趣的severity创建一个新的绑定。

String queueName = channel.queueDeclare().getQueue();

for(String severity : argv){    
  channel.queueBind(queueName, EXCHANGE_NAME, severity);
}

把它们放在一起

python-four.png EmitLogDirect.java类的代码:

public class EmitLogDirect { private static final String EXCHANGE_NAME = "direct_logs"; public static void main(String[] argv) throws java.io.IOException { ConnectionFactory factory = new ConnectionFactory(); factory.setHost("localhost"); Connection connection = factory.newConnection(); Channel channel = connection.createChannel(); channel.exchangeDeclare(EXCHANGE_NAME, "direct"); String severity = getSeverity(argv); String message = getMessage(argv); channel.basicPublish(EXCHANGE_NAME, severity, null, message.getBytes()); System.out.println(" [x] Sent '" + severity + "':'" + message + "'"); channel.close(); connection.close(); } //.. } 

ReceiveLogsDirect.java类的代码:

public class ReceiveLogsDirect { private static final String EXCHANGE_NAME = "direct_logs"; public static void main(String[] argv) throws java.io.IOException, java.lang.InterruptedException { ConnectionFactory factory = new ConnectionFactory(); factory.setHost("localhost"); Connection connection = factory.newConnection(); Channel channel = connection.createChannel(); channel.exchangeDeclare(EXCHANGE_NAME, "direct"); String queueName = channel.queueDeclare().getQueue(); if (argv.length < 1){ System.err.println("Usage: ReceiveLogsDirect [info] [warning] [error]"); System.exit(1); } for(String severity : argv){ channel.queueBind(queueName, EXCHANGE_NAME, severity); } System.out.println(" [*] Waiting for messages. To exit press CTRL+C"); QueueingConsumer consumer = new QueueingConsumer(channel); channel.basicConsume(queueName, true, consumer); while (true) { QueueingConsumer.Delivery delivery = consumer.nextDelivery(); String message = new String(delivery.getBody()); String routingKey = delivery.getEnvelope().getRoutingKey(); System.out.println(" [x] Received '" + routingKey + "':'" + message + "'"); } } } 

如平常那样编译(看指南第一部分,编译和类路径的建议)。为了方便,当我们运行实例是,我们现在使用一个环境变量$CP(在windows环境上是%CP%)表示类路径。 如果你想仅保存warningerror记录不包含info记录信息到文件里,打开一个控制平台并输入:

$ java -cp $CP ReceiveLogsDirect warning error > logs_from_rabbit.log 

如果你想在你的屏幕上看所有的日志信息,打开一个新的终端并键入:

$ java -cp $CP ReceiveLogsDirect info warning error [*] Waiting for logs. To exit press CTRL+C 

例如,为了发布一个错误日志信息,仅需要键入:

$ java -cp $CP EmitLogDirect error "Run. Run. Or it will explode." [x] Sent 'error':'Run. Run. Or it will explode.' 

EmitLogDirect.java sourceReceiveLogsDirect.java source的所有源代码。

阅览指南第五部分,查看如何根据一个模式来监听消息。

相关实践学习
消息队列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
相关文章
|
2月前
|
消息中间件 Java Kafka
RabbitMQ 入门
RabbitMQ 入门
|
5月前
|
消息中间件 新零售 弹性计算
云消息队列 RabbitMQ 版入门训练营,解锁对比开源优势与零基础实战
欢迎加入「云消息队列 RabbitMQ 版入门训练营」。
171 16
|
4月前
|
消息中间件 存储 Java
分享一下rocketmq入门小知识
分享一下rocketmq入门小知识
57 0
分享一下rocketmq入门小知识
|
4月前
|
网络协议 物联网 测试技术
App Inventor 2 MQTT拓展入门(保姆级教程)
本文演示的是App和一个测试客户端进行消息交互的案例,实际应用中,我们的测试客户端可以看着是任意的、支持MQTT协议的硬件,通过订阅及发布消息,联网硬件与我们的App进行双向数据通信,以实现万物互联的智能控制效果。
219 2
|
5月前
|
消息中间件 Kafka
面试题Kafka问题之RabbitMQ的路由配置工作如何解决
面试题Kafka问题之RabbitMQ的路由配置工作如何解决
68 1
EMQ
|
7月前
|
安全 网络性能优化
MQTT 5.0 报文(Packets)入门指南
MQTT 控制报文是 MQTT 数据传输的最小单元。MQTT 客户端和服务端通过交换控制报文来完成它们的工作,比如订阅主题和发布消息。
EMQ
726 14
MQTT 5.0 报文(Packets)入门指南
|
6月前
|
消息中间件 存储 Kafka
01.RabbitMQ入门
01.RabbitMQ入门
59 0
|
7月前
|
消息中间件 监控 Linux
RabbitMQ轻松入门:从零开始的部署与安装指南
RabbitMQ轻松入门:从零开始的部署与安装指南
143 0
RabbitMQ轻松入门:从零开始的部署与安装指南
|
6月前
|
消息中间件 存储 前端开发
RabbitMQ在Java中的完美实现:从入门到精通
本文由木头左介绍如何在Java项目中使用RabbitMQ。RabbitMQ是开源的AMQP实现,支持多种客户端,适合分布式系统中的消息传递。首先需安装Erlang和RabbitMQ,接着在Java项目中添加RabbitMQ客户端库依赖。通过创建连接工厂和连接,建立与RabbitMQ的通信,并展示了创建连接和通道的代码示例。
|
7月前
|
消息中间件 Docker 微服务
RabbitMQ入门指南(十一):延迟消息-延迟消息插件
RabbitMQ是一个高效、可靠的开源消息队列系统,广泛用于软件开发、数据传输、微服务等领域。本文主要介绍了DelayExchange插件、延迟消息插件实现延迟消息等内容。
1029 0