上一篇中我们构建了一个简单的日志系统，我们可以把日志消息广播给多个接受者。
这篇中我们将来添加一个特性只接收部分消息。例如我只将一些错误log存到文件中，把所有的log都打印到控制台里。
1、绑定（Bindings）
在上篇博文中，我们已经创建了一个binding，代码如下：
channel.queueBind(queueName, EXCHANGE_NAME, "");
一个binding就是exchange和Queue之间的一个关系。可以简单的理解为：这个Queue对其相对于的exchange的消息感兴趣（原文是the queue is interested in messages from this exchange，能理解什么意思，但总觉得怪怪的）。
Binding可以使用一个已经存在的routingKey参数。为了避免和basic_publish参数混淆，我们称之为binding key。下边就是我们怎么用key来创建一个binding：
channel.queueBind(queueName, EXCHANGE_NAME, "black");
binding key的意义有时候取决于exchange的类型。对于Fanout类型的exchange，会忽略binding key。
2、Direct类型的exchange
我们上篇博文中的日志系统会把所有的log消息广播给所有的消费者。我们想扩展来根据他们的日志级别来过滤log消息。例如：我们只想把error级别的日志写到磁盘文件中，而其它级别的日志消息则过滤掉。
我们之前使用的fanout类型的exchange，但这样就不会有太多的灵活性。
在这里我们将要使用direct类型的exchange。Direct类型exchange的路由算法是很简单的：要想一个消息能到达这个队列，需要binding key和routing key正好能匹配得上。
为了说明这个道理，可以看看下边的描述：
image
在这样的结构中，我们可以看到direct类型的exchange X，有两个queue绑定到它。第一个queue是以orange为binding key绑定到exchange X上的，第二个queue是由两个binding key（black和green）绑定到exchange X的。
在这样的设置中，一条消息被推送到exchange，如果使用的routing key是orange，那么消息就会被路由到Q1中；如果使用的routing key是black或green，那么该消息将会被路由到Q2中。其它的消息都将会被丢弃掉。
3、多重绑定（Multiple bindings）
image
用同一个binding来把多个queue绑定到同一个exchange也是可行的。例如在之前例子的基础上，在X和Q1之间添加binding key名字为black，这样的话，这里的direct类型的exchange就和fanout类型的一样了，可以把消息推送给所有的queue。带有routing key为black的消息将会被推送到Q1和Q2中。
4、发送日志（Emitting logs）
我们将会使用这种模型，不使用fanout类型的exchange，而是使用direct类型的。我们使用日志级别做为routing key，接收端根据设置的日志级别做为binding key来接收消息。首先来看看发射日志：
如之前一样，首先来创建一个exchange：
channel.exchangeDeclare(EXCHANGE_NAME, "direct");
然后准备发送消息;
channel.basicPublish(EXCHANGE_NAME, severity, null, message.getBytes());
这里的”severity”可以是”info”、“warning”、”error”等。
5、订阅（Subscribing）
这里接收消息和上篇博文中的一样，只是有一点例外：我们将会为每一个感兴趣的日志级别进行绑定。
String queueName = channel.queueDeclare().getQueue();
for(String severity : argv){
channel.queueBind(queueName, EXCHANGE_NAME, severity);
}
6、最终实现
- EmitLogDirect.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();
//声明direct类型的exchange
channel.exchangeDeclare(EXCHANGE_NAME, "direct");
//拿到日志级别
String severity = getSeverity(argv);
//拿到日志消息
String message = getMessage(argv);
//指定routing key，发送消息
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();
//声明direct类型的exchange
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 + "'");
}
}
}
- 运行三个日志接收器：
接收error和info级别的日志：
image
接收error级别的日志：
image
接收info级别的日志：
image
- 运行两个日志发生器：
产生error级别的日志：
image.png
产生info级别的日志：
image.png
- 观察接收器端的变化：
接收error级别的接收器，只接收error级别的日志：
image.png
接收info级别的接收器，只接收info级别的日志：
image.png
Error和info级别日志都接收的接收器，info和error级别的日志都接收：
image.png
7、总结：
要记住生产者端的routing key，那么在消费者端设置binding key和之前的routing key一样，就可以用direct类型的exchange了，以此来获取到自己需要的消息。
参考链接：http://www.rabbitmq.com/tutorials/tutorial-four-java.html
网友评论