路由(routing)
在上一个教程中，我们实现了一个简单的日志系统。我们将日志消息广播到很多个消费者。
在这个教程我们将给它加一个特性 —— 我们将使它可以只订阅消息的一个子集。例如：我们直接将致命的错误信息打印到日志(保存在磁盘中)，同时能将所有的日志信息打印在控制台上。
绑定(Bindings)
在前面的教程中我们已经用到了绑定，可以重新调用那段代码：
channel.queueBind(queueName, EXCHANGE_NAME, "");
一个exchange和queue之间的绑定关系。换句话说：这个queue对exchange中的消息感兴趣。
绑定需要增加一个额外的参数routingKey。为了避免和推送消息中basic_publish的参数名造成混乱，消费者中我们叫它binding key。下面展示创建一个binding key:
channel.queueBind(queueName, EXCHANGE_NAME, "black");
binding key的关键在于依赖的exchange类型。类型为fanout的exchange，上个教程用到的，忽略了它的值。
直接交换(direct exchange)
上个教程我们的日志系统直接将所有的信息广播给所有的消费者。我们想在这个的基础上根据消息的严重程度进行过滤。例如：我们想有一个程序用来将重要的错误信息存储到磁盘当中，而不像浪费空间去存储warning或者是info级别的日志消息。
我们使用的fanout类型的exchange，并不能给我们提供这样的灵活性 —— 它只能盲目的进行广播。
我们将用direct类型的exchange来代替它。direct exchange背后的路由算法很简单 —— 生产者的routing key 完全匹配消费者中的binding key。
为了说明这些，请看下图中的配置：
rabbit-direct
在上图的配置中，我们看到有两个队列绑定了类型为direct的exchange X。第一个队列绑定了的key为orange，第二个队列有两个绑定，一个绑定的key为black，另一个是green。
在这样的一个配置中，推送到exchange的消息，routing key为orange的消息将路由到队列1(Q1)中，routing key为black或者为green的消息将路由到队列2(Q2)中。其他的消息将被丢弃。
多个绑定(multiple bindings)
多个队列绑定相同的key是完全允许的。在我们的例子当中，我们可以在X和Q1之间添加一个绑定key为black的关系。这样，类型为direct的exchange就类似于fanout了，能将消息广播到所有匹配的队列中。当消息的routing key为black时，将分发到Q1和Q2中。
发送日志(emitting logs)
我们将在我们的日志系统中使用这个模式。将消息发送到类型为direct的exchange中，而不是fanout类型的exchange。这样接收程序就能选择重要的消息接收了。
像以前一样，我们需要创建一个exchange先：
channel.exchangeDeclare(EXCHANGE_NAME, "direct");
然后发送一个消息：
channel.basicPublish(EXCHANGE_NAME, severity, null, message.getBytes());
简单起见，我们将严重的级别的定义为：info、warning、error。
订阅(subscribing)
接收消息将和前面的教程差不多，但有一点除外 —— 我们将给我们所有感兴趣的每种严重的消息创建绑定关系。
String queueName = channel.queueDeclare().getQueue();
for(String severity : argv){
channel.queueBind(queueName, EXCHANGE_NAME, severity);
}
信息汇总
rabbitmq-routing
官网的使用命令行执行的。这里我们将一次性的向exchange发送6条消息，info、warn、error三个级别各两条。如上图，我们创建两个消费者，这里我们创建两个类，一个接收routing key为error的消息，并将其打印到文件中；另一个接收所有消息并打印到控制台。
package com.roachfu.tutorial.rabbitmq.website.direct;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
public class EmitLogDirect {
private static final String EXCHANGE_NAME = "direct.log";
public static void main(String[] args) throws IOException, TimeoutException {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();
channel.exchangeDeclare(EXCHANGE_NAME, "direct");
System.out.println(" [*] begin sent message to exchange");
/* 分别发送两条 info,warn,error基本的消息 */
channel.basicPublish(EXCHANGE_NAME, "error", null, "[error] - this is first error message.".getBytes("UTF-8"));
channel.basicPublish(EXCHANGE_NAME, "warn", null, "[warn ] - this is first warn message.".getBytes("UTF-8"));
channel.basicPublish(EXCHANGE_NAME, "info", null, "[info ] - this is first info message.".getBytes("UTF-8"));
channel.basicPublish(EXCHANGE_NAME, "error", null, "[error] - this is second error message.".getBytes("UTF-8"));
channel.basicPublish(EXCHANGE_NAME, "warn", null, "[warn ] - this is second warn message.".getBytes("UTF-8"));
channel.basicPublish(EXCHANGE_NAME, "info", null, "[info ] - this is second info message.".getBytes("UTF-8"));
System.out.println( " [x] done. . . ");
channel.close();
connection.close();
}
}
package com.roachfu.tutorial.rabbitmq.website.direct;
import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
public class ReceiveLogDirectToConsole {
private static final String EXCHANGE_NAME = "direct.log";
public static void main(String[] args) throws IOException, TimeoutException {
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();
channel.queueBind(queueName, EXCHANGE_NAME, "info");
channel.queueBind(queueName, EXCHANGE_NAME, "warn");
channel.queueBind(queueName, EXCHANGE_NAME, "error");
System.out.println(" [*] Waiting for message and handle it to console . . . ");
Consumer consumer = new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
String message = new String(body, "UTF-8");
System.out.println(" [x] receive " + envelope.getRoutingKey() + " : '" + message + "'");
}
};
channel.basicConsume(queueName,consumer);
}
}
package com.roachfu.tutorial.rabbitmq.website.direct;
import com.rabbitmq.client.*;
import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
public class ReceiveLogDirectToFile {
private static final String EXCHANGE_NAME = "direct.log";
public static void main(String[] args) throws IOException, TimeoutException {
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();
channel.queueBind(queueName, EXCHANGE_NAME, "error");
System.out.println(" [*] Waiting for message and handle it to file . . . ");
Consumer consumer = new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
File file = new File("/temp/direct.log");
FileOutputStream out = new FileOutputStream(file, true);
out.write(body);
out.write(("\r\n").getBytes());
out.flush();
out.close();
}
};
channel.basicConsume(queueName,consumer);
}
}
我们先运行两个消费者，然后运行生产者。看输出结果：
ReceiveLogDirectToConsole 消费者
rabbitmq-direct-console
ReceiveLogDierctToFile 消费者
rabbitmq-direct-file
网友评论