注：这是RabbitMQ-java版Client的指导教程翻译系列文章，欢迎大家批评指正
第一篇Hello Word了解RabbitMQ的基本用法
第二篇Work Queues介绍队列的使用
第三篇Publish/Subscribe介绍转换器以及其中fanout类型
第四篇Routing介绍direct类型转换器
第五篇Topics介绍topic类型转换器
第六篇RPC介绍远程调用
在上一篇指导教程中，我们创建了一个日志系统，可以把日志消息广播给很多接受者。
在这篇指导教程中，我们需要添加一个功能：可以订阅消息的一部分。例如：我们会直接将严重的错误信息生成日志文件（保存在空余的磁盘上），但是依然会把所有的日志信息显示在控制台。
绑定(Bindings)
在上篇指导教程的例子中，我们已经创建过绑定的实例，你可能会觉得跟下面的代码类似：
channel.queueBind(queueName, EXCHANGE_NAME, "");
绑定的含义是转换器和队列之间的一种关联，通俗来说就是一个队列对这个转换器中的消息感兴趣。
绑定可以带有一个参数：routingKey。为了避免和basic_publish中的参数产生困惑，我们将这个参数叫着binding key(绑定钥匙)，下面是我们创建一个带有钥匙的绑定。
channel.queueBind(queueName, EXCHANGE_NAME, "black");
这个绑定钥匙的意思取决于转换器的类型，如果是我们之前使用的fanout类型转换器，那么会忽略绑定钥匙的意义。
直接转换器(Direct exchange)
在上篇指导教程中，我们的日志系统会广播消息给所有绑定转换器的消费者。现在我们扩展一下：根据消息的级别来过滤消息。举例来说，我们想一个应用只接受严重级别的消息并且写入到磁盘里，就不用浪费磁盘空间去保存警告或者信息日志的消息。
如果使用fanout转换器，那样就没有什么灵活性，不停的愚蠢的广播。
可以使用direct转换器，它的路由选择的算法是容易理解，一个消息之所以到这个队列中去，是因为队列的binding Key和发出消息的routingkey相匹配。
为了说明这个问题，看下下面的结构：
direct-exchange.png
这张结构图中，可以看到有两个队列绑定着类型为direct的转换器，第一个队列绑定钥匙为orange，第二个绑定钥匙有两个：一个是black另一个是green。
在上面的结构图中，一个带有routingkey为orange的消息发送给转换器将会被发送到队列Q1中，带有routing Key为black和green将会被发送给到队列Q2中，其他所有的消息将会被清除。
多重绑定(Multiple bindings)
direct-exchange-multiple.png
多个队列拥有相同的binding key是完全合规的，上图中我们可以在转换器x和带有bindingkey为black的队列Q1建立绑定关系。在这种情况下，direct类型的转换器具有fanout类型的一样特性，可以广播给所有匹配的队列消息。一个routingkey为black的消息将会被发送到Q1和Q2两个队列中。
发送消息(Emitting logs)
我们使用这种模型应用到日志系统上，发送给消息给direct而不是fanout类型的转换器。将以日志严重等级作为routing key。按照那种方式，消费者应用将会选择接受日志的严重等级的消息。首先我们先发送消息。
总是一样的，先声明一个转换器：
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);
}
综合
python-four.png
下面是EmitLogDirect.java类，这里下载：
import com.rabbitmq.client.*;
import java.io.IOException;
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类，这里下载：
import com.rabbitmq.client.*;
import java.io.IOException;
public class ReceiveLogsDirect {
private static final String EXCHANGE_NAME = "direct_logs";
public static void main(String[] argv) throws Exception {
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");
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] Received '" + envelope.getRoutingKey() + "':'" + message + "'");
}
};
channel.basicConsume(queueName, true, consumer);
}
}
跟以前一样编译，运行的时候为了方面，我们使用环境便来个$CP作为路径配置：
javac -cp $CP ReceiveLogsDirect.java EmitLogDirect.java
如果你想把warning和error(而不是info)类型的日消息保存到文件中，只需要打开一个控制台和记录：
java -cp $CP ReceiveLogsDirect warning error > logs_from_rabbit.log
如果你想在你的屏幕上看到所有的日志消息，新打开一个终端和查看就可以：
java -cp $CP ReceiveLogsDirect info warning error
# => [*] Waiting for logs. To exit press CTRL+C
举个例子，发送一个error类型的日志消息：
java -cp $CP EmitLogDirect error "Run. Run. Or it will explode."
# => [x] Sent 'error':'Run. Run. Or it will explode.'
第四节的内容大致翻译完了，这里是原文链接。接着进入下一节：Topics。
终篇是我对RabbitMQ使用理解的总结文章，欢迎讨教。
--谢谢--
网友评论