3、路由选择 (Routing)

本篇博客我们准备给日志系统添加新的特性,让日志接收者能够订阅部分消息。例如,我们可以仅仅将致命的错误写入日志文件,然而仍然在控制面板上打印出所有的其他类型的日志消息。

1、绑定(Bindings)

在上一篇博客中我们已经使用过绑定。类似下面的代码:
channel.queueBind(queueName, EXCHANGE_NAME, "");
绑定表示转发器与队列之间的关系。也可以简单的认为:队列对该转发器上的消息感兴趣。
绑定可以附带一个额外的参数routingKey。为了与避免basicPublish方法(发布消息的方法)的参数混淆,我们准备把它称作绑定键(binding key)。下面展示如何使用绑定键(binding key)来创建一个绑定:
channel.queueBind(queueName, EXCHANGE_NAME, "black");
绑定键的意义依赖于转发器的类型。对于fanout类型,忽略此参数。

2、直接转发(Directexchange)

上一篇的日志系统广播所有的消息给所有的消费者。我们希望可以对其扩展,来允许根据日志的严重性进行过滤日志。例如:我们可能希望把致命类型的错误写入硬盘,而不把硬盘空间浪费在警告或者消息类型的日志上。
    之前我们使用fanout类型的转发器,但是并没有给我们带来更多的灵活性:仅仅可以愚蠢的转发。
    我们将会使用direct类型的转发器进行替代。direct类型的转发器背后的路由转发算法很简单:消息会被推送至绑定键(binding key)和消息发布附带的选择键(routing key)完全匹配的队列。
图解:

上图,我们可以看到direct类型的转发器与两个队列绑定。第一个队列与绑定键orange绑定,第二个队列与转发器间有两个绑定,一个与绑定键black绑定,另一个与green绑定键绑定。

这样的话,当一个消息附带一个选择键(routing key) orange发布至转发器将会被导向到队列Q1。消息附带一个选择键(routing key)black或者green将会被导向到Q2.所有的其他的消息将会被丢弃。

3、多重绑定(multiplebindings)

使用一个绑定键(binding key)绑定多个队列是完全合法的。如上图,一个附带选择键(routing key)的消息将会被转发到Q1和Q2。

4、发送日志(Emittinglogs)

我们准备将这种模式用于我们的日志系统。我们将消息发送到direct类型的转发器而不是fanout类型。我们将把日志的严重性作为选择键(routing key)。这样的话,接收程序可以根据严重性来选择接收。我们首先关注发送日志的代码:

像以前一样,我们需要先创建一个转发器:

channel.exchangeDeclare(EXCHANGE_NAME,"direct");

然后我们准备发送一条消息:

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

为了简化代码,我们假定‘severity’是‘info’,‘warning’,‘error’中的一个。

5、订阅

接收消息的代码和前面的博客的中类似,只有一点不同:我们给我们所感兴趣的严重性类型的日志创建一个绑定。

for(String severity: argv) {

channel.queueBind(queueName, EXCHANGE_NAME, severity);

}

6、完整的实例

发送端

public class producer5 {private static final String EXCHANGE_NAME = "ex_logs_direct";private static final String[] SEVERITIES = { "info", "warning", "error" };  public static void main(String[] argv) throws Exception { // 创建连接和频道  ConnectionFactory factory = new ConnectionFactory();  factory.setHost("localhost");  Connection connection = factory.newConnection();  Channel channel = connection.createChannel();  // 声明转发器和类型  channel.exchangeDeclare(EXCHANGE_NAME, "direct");//发送3条消息  for(int i=0; i<3; i++){  String severity = SEVERITIES[i];String message = severity + "_log :" + UUID.randomUUID().toString();  // 发布消息至转发器,指定routingkey  channel.basicPublish(EXCHANGE_NAME, severity, null, message.getBytes());   System.out.println(" [x] Sent '" + message + "'");  }  channel.close();  connection.close();  }  }

接收端1

public class consumer5_1 {private static final String EXCHANGE_NAME = "ex_logs_direct";  private static final String[] SEVERITIES = { "info", "warning", "error" };  public static void main(String[] argv) throws Exception{ // 创建连接和频道  ConnectionFactory factory = new ConnectionFactory();  factory.setHost("localhost");  Connection connection = factory.newConnection();  Channel channel = connection.createChannel();  // 声明direct类型转发器  channel.exchangeDeclare(EXCHANGE_NAME, "direct");  String queueName = channel.queueDeclare().getQueue();  String severity = SEVERITIES[2];//error类型的日志// 指定binding_key  channel.queueBind(queueName, EXCHANGE_NAME, severity);  System.out.println(" [*] Waiting for "+severity+" logs. 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());  System.out.println(" [x] Received '" + message + "'");  }  }  }

接收端1 binding_key为error,只接收rout ing_key为error的消息

接收端2

public class consumer5_2 {private static final String EXCHANGE_NAME = "ex_logs_direct";  private static final String[] SEVERITIES = { "info", "warning", "error" };  public static void main(String[] argv) throws Exception{ // 创建连接和频道  ConnectionFactory factory = new ConnectionFactory();  factory.setHost("localhost");  Connection connection = factory.newConnection();  Channel channel = connection.createChannel();  // 声明direct类型转发器  channel.exchangeDeclare(EXCHANGE_NAME, "direct");  String queueName = channel.queueDeclare().getQueue();  String severity0 = SEVERITIES[0];String severity1 = SEVERITIES[1];String severity2 = SEVERITIES[2];//绑定三种类型的channel.queueBind(queueName, EXCHANGE_NAME, severity0);// 指定binding_key  channel.queueBind(queueName, EXCHANGE_NAME, severity1);// 指定binding_key  channel.queueBind(queueName, EXCHANGE_NAME, severity2);// 指定binding_key  System.out.println(" [*] Waiting for info、warning、errologs. 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());  System.out.println(" [x] Received '" + message + "'");  }  }  }

接收端2 queueBind()bind了三种routing_key的消息,所以都会接收到

测试:

发送端:

[x] Sent 'info_log:3b758831-60f1-4de6-a610-cb0c9afc46ef'

[x] Sent 'warning_log:c0d71453-f3e3-4b28-84ad-607200ec8c2f'

[x] Sent 'error_log:58a69784-d208-472d-81c2-b55dc7f249d6'

接收端1::

[*] Waiting for error logs. To exitpress CTRL+C

[x] Received 'error_log :58a69784-d208-472d-81c2-b55dc7f249d6'

接收端2:

[*] Waiting for info、warning、errologs. Toexit press CTRL+C

[x] Received 'info_log:3b758831-60f1-4de6-a610-cb0c9afc46ef'

[x] Received 'warning_log:c0d71453-f3e3-4b28-84ad-607200ec8c2f'

[x] Received 'error_log :58a69784-d208-472d-81c2-b55dc7f249d6'

原文:http://blog.csdn.net/lmj623565791/article/details/37669573

RabbitMQ-三、Java使用--3、路由选择 (Routing)相关推荐

  1. GPS定位系统(三)——Java后端

    前言 GPS系列--Java端,github项目地址 前面已经了解或者实现了Android端的gps上传定位信息,现在就差后台的接口支持了. 我们需要数据库来储存上传的定位信息,并且还要满足不同的人的 ...

  2. RabbitMQ (四) 路由选择 (Routing)

    上一篇博客我们建立了一个简单的日志系统,我们能够广播日志消息给所有你的接收者,如果你不了解,请查看:RabbitMQ (三) 发布/订阅.本篇博客我们准备给日志系统添加新的特性,让日志接收者能够订阅部 ...

  3. Java笔记-使用RabbitMQ的Java接口实现Routing(路由模式)

    目录 基本概念 代码与实例 基本概念 过程图如下: 主要是把交换机设置为直连的方式direct直连的方式然后把 数据 发送给交换机. 交换机再通过路由的Key值转发到队列上. 每一个客户端,都有一个队 ...

  4. rabbitMQ实例(java)

    文章转载自:https://www.cnblogs.com/stormli/p/rabbitmq.html RabbitMQ 一.背景 RabbitMQ是一个由erlang开发的AMQP(Advanc ...

  5. RabbitMQ指南之四:路由(Routing)和直连交换机(Direct Exchange)

    在上一章中,我们构建了一个简单的日志系统,我们可以把消息广播给很多的消费者.在本章中我们将增加一个特性:我们可以订阅这些信息中的一些信息.例如,我们希望只将error级别的错误存储到硬盘中,同时可以将 ...

  6. RabbitMQ(三) HelloWorld 单生产者单消费者示例实现

    一.创建Maven工程,引入RabbitMQ依赖. pom.xml 如下: <?xml version="1.0" encoding="UTF-8"?&g ...

  7. java 消息队列_消息队列-RabbitMQ在JAVA中的应用(1)

    一.环境准备 安装RabbitMq 安装可查看另外文章本文就不做介绍了. 本文使用的erl和RabbitMQ版本分别是10.4和3.7.15 二.进入正题 1.创建SpringBoot项目 1.1.生 ...

  8. AMQP 协议及 RabbitMQ 的 Java 用例

    AMQP 协议 AMQP is the Internet Protocol for Business Messaging (AMQP) Version 1.0 AMQP Working Group 0 ...

  9. 大三Java后端暑期实习面经总结——JVM篇

    博主现在大三在读,从三月开始找暑期实习,暑假准备去tx实习啦!总结下了很多面试真题,希望能帮助正在找工作的大家!相关参考都会标注原文链接,尊重原创! 目录 1. jvm体系结构 2. 类加载器 3. ...

最新文章

  1. NBT:牛瘤胃微生物组的参考基因组集
  2. echart旭日图_150Echarts - 旭日图(Sunburst Label Rotate)
  3. lua.c:80:31: fatal error: readline/readline.h: No such file or directory
  4. SAP CRM Fiori My note应用 add to -append Frontend implementation
  5. 几何画板自定义工具_几何画板sketchpad下载-几何画板sketchpad免费版下载v5.0.6.5
  6. ARP欺骗***技术及防护
  7. 12.第二节 SAPI概述
  8. 「 LaTex 」写论文,natbib宏的参考文献引用格式详解
  9. 联想 m73 黑苹果 软路由 esxi AIO
  10. 持久层框架JPA与Mybatis该如何选型
  11. 更换app图标和名称
  12. 前端获取北京时间_分享js获取标准北京时间的代码 JS如何获取北京时间JS
  13. 爱情不是等你有空才珍惜的
  14. 计算机二级word家长回执单得分,全国计算机国家二级office课件_word第一次课.doc...
  15. 关于http的Authorization
  16. 用transform:scale();缩放
  17. 万门大学MySQL特训班_如何评价万门大学「理论物理一月特训班」?
  18. 获取免费代理IP,并检验IP有效性
  19. linux 查看磁盘、文件夹、文件大小(df du)
  20. Compose Multiplatform结合MVI模式--初步尝试

热门文章

  1. 使用JSP/Servlet技术开发新闻发布系统
  2. 为了更好的实践与应用-机器人与ROS教学与培训
  3. [Python爬虫] Selenium实现自己主动登录163邮箱和Locating Elements介绍
  4. jquery实现手风琴展开效果
  5. maven如何排除依赖
  6. Q_INVOKABLE与invokeMethod用法详解
  7. 大学生学习摄影如何进行构图?
  8. Windows 7 新功能选集
  9. typora+百度云盘+markor实现多端云同步
  10. c语言 文件编码转换为字符串,c语言下汉字转换(字符串改为utf-8编码)