博客
关于我
强烈建议你试试无所不能的chatGPT,快点击我
RabbitMQ学习总结(6)——消息的路由分发机制详解
阅读量:5969 次
发布时间:2019-06-19

本文共 4094 字,大约阅读时间需要 13 分钟。

hot3.png

 

一、Routing(路由) (using the Java client)

在前面的学习中,构建了一个简单的日志记录系统,能够广播所有的日志给多个接收者,在该部分学习中,将添加一个新的特点,就是可以只订阅一个特定的消息源,也就是说能够直接把关键的错误日志消息发送到日志文件保存起来,不重要的日志信息文件不保存在磁盘中,但是仍然能够在控制台输出,那么这便是我们这部分要学习的消息的路由分发机制。

二、Bindings(绑定)

在前面的学习中已经创建了绑定(bindings),代码如下:
channel.queueBind(queueName, EXCHANGE_NAME, "");

一个绑定就是一个关于exchange和queue的关系,它可以简单的被理解为:队列是从这个exchange中获取消息的。

绑定可以采取一个额外的routingKey的参数,为了避免与basicPublish参数冲突,称之为一个绑定Key,这是如何创建一个带routingKey的绑定的关键。

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

一个绑定Key依赖于exchange的类型,像之前使用fanout类型的exchange,完全忽略了该绑定key的值。

三、Direct exchange(直接交换机)

前面实现的日志记录系统中广播所有的消息给所有的消费者,现在对其进行扩展,允许根据信息的严重程度来对消息进行过滤,比如,希望一个程序写入到磁盘的日志消息只接收错误的消息,而不是浪费磁盘保存所有的日志消息。

为了实现这个目标,使用一个fanout类型的exchange,显然是不能够满足这样的需求的,因为它只能广播所有的消息。

为此将使用一个direct exchange来代替fanout exchange,direct exchange使用简单的路由算法,将消息通过绑定的Key匹配将要到达的队列。

从上面的结构图中可以看出direct exchange X绑定着两个queue(Q1,Q2),第一个queue绑定的routingKey为orange,第二个有两个routingKey被绑定,一个routingKey为black,另外一个routingKey为green.

说明:发送带有routingKey为orange的消息到X(exchange)中,X将该消息路由到Q1中,发送带有routingKey为black和green的消息都将被路由到Q2中,其他所有消息将会被丢弃。

四、Multiple bindings(多绑定)

多个队列绑定相同的routingKey是允许的,在上述实例中,可以把X和Q1用routingKey:black绑定起来,这种情况下,direct exchange将像fanout类型的exchange一样会将消息广播都到所有匹配的queues中,即一个routingKey为black的消息将会被发送到Q1和Q2中。

五、Emitting logs(发送的日志)

使用direct代替fanout类型的exchange,发送消息到一个direct exchange中,将根据消息的重要程度作为routingKey,这样接收程序能够选择它想要接收的日志信息,首先必须先创建一个exchange.
channel.exchangeDeclare(EXCHANGE_NAME, "direct");

其次,发送一条信息:
channel.basicPublish(EXCHANGE_NAME, severity, null, message.getBytes());
为了简化程序,将severity设定为info、warning、error三种类型中的一种。

六、Subscribing(订阅消息)

接收者根据自己感兴趣的severity来创建一个新到的绑定。
String queueName = channel.queueDeclare().getQueue();for(String severity : argv){
channel.queueBind(queueName, EXCHANGE_NAME, severity);}

七、Putting it all together(代码实现)

EmitLogDirect.java代码清单如下:
[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(%CP%在Windows上)的运行时类路径的例子。如果你只想保存 “警告”和“错误”(而不是“信息”)日志消息到一个文件,打开一个控制台和type:

[*] Waiting for logs. To exit press CTRL+C

[x] Sent 'error':'Run. Run. Or it will explode.'
$ java -cp $CP ReceiveLogsDirect warning error > logs_from_rabbit.log
$ java -cp $CP ReceiveLogsDirect info warning error
$ java -cp $CP EmitLogDirect error "Run. Run. Or it will explode."

转载于:https://my.oschina.net/zhanghaiyang/blog/595073

你可能感兴趣的文章
bootloader功能介绍/时钟初始化设置/串口工作原理/内存工作原理/NandFlash工作原理...
查看>>
iOS开发UI篇—Quartz2D使用(矩阵操作)
查看>>
C++ 构造函数与析构函数
查看>>
定时压缩log日志文件
查看>>
秋无痕 Windows XPSP3 集成安装增强版 V201306
查看>>
IT男成都租房记
查看>>
博为峰JavaEE技术文章 ——MyBatis Provider之@SelectProvider SQL方法
查看>>
Java核心API -- 9(异常)
查看>>
apache 编译报错:undefined reference to `apr_array_clear'
查看>>
图像识别DM8127开发攻略——UBOOT的移植说明
查看>>
ubuntu 下升级docker版本
查看>>
EXSi5.5安装篇
查看>>
开始记录吧
查看>>
windows下用php开发类似百度文库应用需要的工具和问题
查看>>
css模拟select设置高度在ie67下有效(也可作为去除边框)
查看>>
互联网思维
查看>>
ecshop备份数据 ecshop转移数据 ecshop更换主机
查看>>
手机将与瘦客户机争夺办公桌面
查看>>
ubuntu下针对php的thrift 安装折腾记录
查看>>
使用C#客户端访问FTP服务的一个解决方案
查看>>