5.Topics#前山翻译

注：这是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:

* 代表一个单词

# 代表零个或多个单词

例子最容易解释清楚：

python-five.png

上述例子中，我们发送全是描述动物的消息，这些消息的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使用理解的总结文章，欢迎讨教。
--谢谢--