为什么80%的码农都做不了架构师?>>>   

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

1. 绑定(Bindings)

在上一篇博客中我们已经使用过绑定。类似下面的代码:

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

绑定表示转发器与队列之间的关系。我们也可以简单的认为:队列对该转发器上的消息感兴趣。

绑定可以附带一个额外的参数routingKey。为了避免与basicPublish方法(发布消息的方法)的参数混淆,我们准备把它称作绑定键(Binding key)。下面展示如何使用绑定键(binding key)来创建一个绑定:

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

绑定键的意义依赖于转发器的类型。对于fanout类型,忽略此参数。

2. 直接转发(Direct exchange)

上一篇的日志系统广播所有的消息给所有的消费者。我们希望可以对其扩展,来允许根据日志的严重性进行过滤日志。例如:我们可能希望把致命类型的错误写入硬盘,而不把硬盘空间浪费在警告或者消息类型的日志上。之前我们使用fanout类型的转发器,但是并没有给我们带来更多的灵活性,仅仅可以愚蠢的转发。

我们将会使用direct类型的转发器进行替代。direct类型的转发器背后的路由转发算法很简单:消息会被推送至绑定键(binding key)和消息发布附带的选择键(routing key)完全匹配的队列。图解:

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

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

3. 多重绑定(multiple bindings)

使用一个绑定键(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. 订阅

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

String queueName = channel.queueDeclare().getQueue();
for(String severity : args){channel.queueBind(queueName,EXCHANGE_NAME,severity);
}

6. 完整的实例

发送端:EmitLogDirect.java:

package cc.openscanner;import java.util.Random;
import java.util.UUID;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;public class EmitLogDirect {private static final String EXCHANGE_NAME = "ex_logs_direct";private static final String[] SEVERITIES = {"info","warning","error"};public static void main(String[] args) throws java.io.IOException {//创建连接和通道ConnectionFactory factory = new ConnectionFactory();factory.setHost("localhost");Connection connection = factory.newConnection();Channel channel = connection.createChannel();//声明转发器的类型channel.exchangeDeclare(EXCHANGE_NAME, "direct");//发送6条消息for(int i=0;i<6;i++){String severity = getSeverity();String message = severity + "_log : " + UUID.randomUUID().toString();//发布消息至转发器,指定routing keychannel.basicPublish(EXCHANGE_NAME, severity, null,message.getBytes());System.out.println(" [x] Sent '" + message + "'");  }channel.close();connection.close();}private static String getSeverity(){Random random = new Random();int ranVal = random.nextInt(3);return SEVERITIES[ranVal];}
}

随机发送6条随机类型(routing key)的日志给转发器。

接收端:ReceivelLogsDirect.java:

package cc.openscanner;import java.util.Random;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.QueueingConsumer;public class ReceiveLogsDirect {private static final String EXCHANGE_NAME = "ex_logs_direct";private static final String[] SEVERITIES = {"info","warning","error"};public static void main(String[] args) throws java.io.IOException,java.lang.InterruptedException{//创建连接和通道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 = getSeverity();//指定binding keychannel.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 + "'");}}private static String getSeverity(){Random random = new Random();int ranVal = random.nextInt(3);return SEVERITIES[ranVal];}
}

接收端随机设置一个日志严重级别(binding key)。

我开启了3个接收端程序,两个准备接收error类型日志,一个接收info类型日志,然后运行发送端程序运行结果:

[x] Sent 'error_log :d142b096-46c0-4380-a1d2-d8b2ac136a9c'
 [x] Sent 'error_log :55ee1fc4-c87c-4e5e-81ba-49433890b9ce'
 [x] Sent 'error_log :d01877d6-87c7-4e0a-a109-697d122bc4c9'
 [x] Sent 'error_log :b42471b1-875c-43f1-b1ea-0dd5b49863f3'
 [x] Sent 'info_log :a6c1bc87-efb0-43eb-8314-8a74c345ed05'
 [x] Sent 'info_log :b6a84b6a-353e-4e88-8c23-c791d93b44be'

------------------------------------------------------------------------------------

[*] Waiting for error logs. To exit press CTRL+C
 [x] Received 'error_log :d142b096-46c0-4380-a1d2-d8b2ac136a9c'
 [x] Received 'error_log :55ee1fc4-c87c-4e5e-81ba-49433890b9ce'
 [x] Received 'error_log :d01877d6-87c7-4e0a-a109-697d122bc4c9'
 [x] Received 'error_log :b42471b1-875c-43f1-b1ea-0dd5b49863f3'

------------------------------------------------------------------------------------

[*] Waiting for error logs. To exit press CTRL+C
 [x] Received 'error_log :d142b096-46c0-4380-a1d2-d8b2ac136a9c'
 [x] Received 'error_log :55ee1fc4-c87c-4e5e-81ba-49433890b9ce'
 [x] Received 'error_log :d01877d6-87c7-4e0a-a109-697d122bc4c9'
 [x] Received 'error_log :b42471b1-875c-43f1-b1ea-0dd5b49863f3'

------------------------------------------------------------------------------------

[*] Waiting for info logs. To exit press CTRL+C
 [x] Received 'info_log :a6c1bc87-efb0-43eb-8314-8a74c345ed05'
 [x] Received 'info_log :b6a84b6a-353e-4e88-8c23-c791d93b44be'

注意,使用同一个binding key绑定多个队列,多个队列都会收到该消息(message),如上例!可以看到,我们实现了博客开头所描述的特性,接收者可以自定义自己感兴趣类型的日志。

其实文章这么长就在说:发送消息时可以设置routing key,接收队列与转发器间可以设置binding key,接收者接收routing key与binding key相同的消息。

转载于:https://my.oschina.net/fhd/blog/376240

RabbitMQ之路由选择相关推荐

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

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

  2. RabbitMQ之主题(Topic)

    为什么80%的码农都做不了架构师?>>>    上一篇博客中,我们进一步改良了我们的日志系统.我们使用direct类型转发器,使得接收者有能力进行选择性的接收日志,而非fanout那 ...

  3. RabbitMQ使用简述

    RabbitMQ基于AMQP协议. AMQP:是一个高级抽象层消息通信协议,RabbitMQ是AMQP协议的实现 RabbitMQ使用:Exchange(交换机)根据routing-key(路由选择键 ...

  4. RabbitMQ系列教程之四:路由(Routing)

    在上一个教程中,我们构建了一个简单的日志系统,我们能够向许多消息接受者广播发送日志消息. 在本教程中,我们将为其添加一项功能 ,这个功能是我们将只订阅消息的一个子集成为可能. 例如,我们可以只将关键的 ...

  5. (转)RabbitMQ学习之主题topic(java)

    http://blog.csdn.net/zhu_tianwei/article/details/40887775 参考:http://blog.csdn.NET/lmj623565791/artic ...

  6. ActiveMQ, RabbitMQ和ZeroMQ 选型关注点

    选择MQ时,主要关注的特性,可能就以下几个: 通信模式(是否满足业务场景): ActiveMQ: queue(producer/consumer), topic(publisher/subsriber ...

  7. RabbitMQ简介及简单使用

    文章目录 1.RabbitMQ基本介绍 2.RabbitMQ简单使用案例 2.1在RabbitMQ平台上创建一个队列 2.2编写生产者代码 2.3编写消费者代码 3.RabbitMQ如何保证消息不丢失 ...

  8. RabbitMQ集群环境搭建

    RabbitMQ集群环境搭建 一.安装基础工具包 二:安装Erlang和RabbitMQ 1. 安装Erlang 2. 安装RabbitMQ 3. 启动RabbitMQ 4. 启动RabbitMQ管理 ...

  9. 【rabbitMQ】01 全面理解rabbitMQ

    随着用户流量的快速增长,由于传统应用在系统接口和服务处理层面仍然沿用高耦合和同步的处理模式,导致接口由于线程阻塞而延长了整体响应时间,即所谓的高延迟,为解决这个问题使用了rabbitMQ. Rabbi ...

最新文章

  1. 利用curl下载文件(进度条显示) 代码片段
  2. Kubernetes — Calico CNI
  3. 怎样使用计算机上的高级共享设置密码,win7系统电脑,如何设置共享,共享计算机的用户名和密码怎样设置...
  4. 选择P2P平台的技巧和方法
  5. 科大星云诗社动态20201227
  6. 删除Windows Service
  7. fullgc触发条件_JVM的内存分配策略以及进入分代的条件
  8. STM32驱动LCD实战
  9. 在Visual C#中用ListView显示数据记录
  10. 前景检测算法(十七)--基于光流算法
  11. Monad B2 For Dotnet Framework 2.0 RC/RTM Released!
  12. android接口类命名规范_Android开发规范
  13. 关于wxwidgets图形界面的关闭窗口的按钮无效的解决办法
  14. 前端Echarts数据可视化
  15. [转自老马的文章]用MODI OCR 21种语言
  16. 股价破300迎新年开门红,投资者们中了苹果的毒?
  17. 微信公众号开发之分享接口实现
  18. STC 32位8051单片机来了
  19. 文献阅读:利用斯托克斯参量验证光波偏振态(补充)
  20. 安装arcgis server10.2时发生错误error1935

热门文章

  1. javascript中的对象之间继承关系
  2. 在Java下连接SQLite数据库
  3. 用OFFICE 2007发送的文章
  4. ruby笔记 基于对象的类(object specific class)
  5. 华为在鸿蒙的另一张王牌,华为的这两张王牌,将助力Mate40系列角逐安卓机皇的位置...
  6. mysql索引b树和hash_B树索引和Hash索引的应用场景和区别(转载)
  7. 微软云 mysql 重启_停止Azure MySQL PaaS数据库
  8. layui select第一次无法选中_Windows 10 无法访问共享的解决办法大全
  9. pvr转png工具_pngtosvg 一个神奇的在线工具
  10. oracle数据泵备份单表,使用数据泵备份/恢复某些表