注:这是RabbitMQ-java版Client的指导教程翻译系列文章,欢迎大家批评指正
第一篇Hello Word了解RabbitMQ的基本用法
第二篇Work Queues介绍队列的使用
第三篇Publish/Subscribe介绍转换器以及其中fanout类型
第四篇Routing介绍direct类型转换器
第五篇Topics介绍topic类型转换器
第六篇RPC介绍远程调用
主题(Topics)
在上篇指导教程中我们改进了日志系统,替换了转换器的类型-fanout只会愚蠢的广播,而使用direct类型,增加选择接受日志的能力。
尽管使用direct类型的转换器提高了我们的日志系统,但是它还有有一些限制:路由选择不能路由多种条件。
在我们的日志系统中,我们不仅想订阅严重等级的日志,还想订阅关于发送日志来源。你可能知道这种来自unix工具系统日志的概念,它路由选择依靠于日志严重等级(info/warn/crit...)和设备(auth/cron/kern...)。
这样就给了我们很大的灵活性,我们可能想监听来自cron设备严重错误的日志和来自ken设备的全部日志。
为了能够在我们的日志系统中实现上述的功能,我们需要学习一个更复杂点类型的转换器:Topics。
主题转换器(Topic exchange)
发送给topic类型的转换器没有唯一的routing_key,它必须是一系列的字符,由点分开,字符可以是任何东西,但是一般会使用详细的特性去关联消息。很少有无用的key,例如:"stock.usd.nyse","nyse.vmw","quick.orange.rabbit"。你可以选择任意个字符,但是不能超过255字节。
绑定的key必须是相同的格式,Topics类型和direct类型的转换器背后逻辑是相似的,发送特殊routing_key的消息将会被分发到拥有相匹配binding key的所有队列中。无论怎么样,这里有两种重要特殊的两种keys:
* 代表一个单词
# 代表零个或多个单词
例子最容易解释清楚:
上述例子中,我们发送全是描述动物的消息,这些消息的routing key将由三个单词组成(两个点分开),第一个单词描述速度,第二个单词描述颜色,第三个单词描述种类:".."。
我们创建了三种绑定方式:Q1绑定key:" .orange. ",Q2绑定key:" ..rabbit "和" lazy.# "。
这些绑定总结如下:
Q1对所有orange颜色的动物感兴趣
Q2想知道所有兔子种类的消息和所有懒惰的动物消息。
routing key为 " quick.orange.rabbit "的消息将会被分发到这两个队列中,routing key为" lazy.orange.elephant "的消息同样会被发送到这两个队列中。另一方面,routingkey为" quick.orange.fox "的消息只会被发送到第一个队列中,routingkey为" lazy.brown.fox "的消息只会被发送到第二个队列中。routingkey为" lazy.pink.rabbit "的消息即使符合一个队列的两个绑定,但还是只发送一条消息到该队列中。key为" quick.brown.fox "的消息没有匹配任何绑定,所以将会被清除。
如果我们不顾任何的协议,发送一个routing key为一个单词或者四个单词的消息,像"orange"或者"quick.orange.male.rabbit",那么会怎么样呢?显然,这样的消息将不会匹配任何绑定所以会被丢失。
但是另一方面,像routing_key为" lazy.orange.male.rabbit "的消息,即使有四个单词,但是匹配Q2队列的绑定,所以会被发送到Q2队列中。
主题转换器说明
主题转换器是非常强大的,并且可以像其他的转换器一样使用。
当一个队列绑定binding key为“#”,它将会接收到所有的消息,不管routing key为什么的消息,这就像是fanout类型的转换器。
当一个队列绑定中不带有字符“ * ” 和“ # ”的时候,主题转换器跟direct转换器是相同的。
综合
我们将主题转换器应用到日志系统中,在routing key为两个单词" . "的日志消息前提下,我们开始工作。
以下的代码和上篇指导文章中的很相似。
下面是EmitLogTopic.java的源代码,这里下载:
import com.rabbitmq.client.*;
import java.io.IOException;
public class EmitLogTopic {
private static final String EXCHANGE_NAME = "topic_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, "topic");
String routingKey = getRouting(argv);
String message = getMessage(argv);
channel.basicPublish(EXCHANGE_NAME, routingKey, null, message.getBytes());
System.out.println(" [x] Sent '" + routingKey + "':'" + message + "'");
connection.close();
}
//...
}
下面是ReceiveLogsTopics.java的源代码,这里下载:
import java.io.IOException;
public class ReceiveLogsTopic {
private static final String EXCHANGE_NAME = "topic_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, "topic");
String queueName = channel.queueDeclare().getQueue();
if (argv.length < 1) {
System.err.println("Usage: ReceiveLogsTopic [binding_key]...");
System.exit(1);
}
for (String bindingKey : argv) {
channel.queueBind(queueName, EXCHANGE_NAME, bindingKey);
}
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);
}
}
编译并且运行程序,在第一篇指导文件中,有windows的路径说明,使用%CP%配置环境变量:
javac -cp $CP ReceiveLogsTopic.java EmitLogTopic.java
接收所有的日志:
java -cp $CP ReceiveLogsTopic "#"
接收设备为kern的日志:
java -cp $CP ReceiveLogsTopic "kern.*"
或者你想接收严重等级的日志:
java -cp $CP ReceiveLogsTopic "*.critical"
你也可以建立多重绑定:
java -cp $CP ReceiveLogsTopic "kern." ".critical"
最后发送routing key为" kern.critical "的消息:
java -cp $CP EmitLogTopic "kern.critical" "A critical kernel error"
希望实践这三个程序的时候感到有趣。请注意关于routing key和binding key是没有任何限制的,你可随意的使用多个routing key参数。
第五节的内容大致翻译完了。这里是原文链接,接着进入下一节:RPC。
终篇是我对RabbitMQ使用理解的总结文章,欢迎讨教。
--谢谢--