RabbitMQ (四) 路由选择 (Routing)

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

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、订阅

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

StringqueueName = channel.queueDeclare().getQueue();

for(Stringseverity : argv)

{

channel.queueBind(queueName, EXCHANGE_NAME, severity);

}

6、完整的实例

发送端:EmitLogDirect.java

package com.zhy.rabbit._04_binding_key;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[] 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");//发送6条消息for (int i = 0; i < 6; i++){String severity = getSeverity();String message = severity + "_log :" + UUID.randomUUID().toString();// 发布消息至转发器,指定routingkeychannel.basicPublish(EXCHANGE_NAME, severity, null, message.getBytes());System.out.println(" [x] Sent '" + message + "'");}channel.close();connection.close();}/** * 随机产生一种日志类型 * * @return */private static String getSeverity(){Random random = new Random();int ranVal = random.nextInt(3);return SEVERITIES[ranVal];}}随机发送6条随机类型(routing key)的日志给转发器~~

接收端:ReceiveLogsDirect.java

package com.zhy.rabbit._04_binding_key;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[] argv) 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 + "'");}}/** * 随机产生一种日志类型 * * @return */private static String getSeverity(){Random random = new Random();int ranVal = random.nextInt(3);return SEVERITIES[ranVal];}}接收端随机设置一个日志严重级别(binding_key)。。。

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

运行结果:

每一个成功者都有一个开始。勇于开始,才能找到成功的路。

RabbitMQ (四) 路由选择 (Routing)

相关文章:

你感兴趣的文章:

标签云: