RabbitMQ入门（4）--路由

AI 运维「开挂」指南 OSC源创会·北京·114期

###路由 ###（使用Java客户端） 在先前的指南中，我们建立了一个简单的日志系统。我们可以将我们的日志信息广播到多个接收者。 在这部分的指南中，我们将要往其中添加一个功能-让仅仅订阅一个消息的子集成为可能。例如，我们可以直接将关键的错误信息指向到日志文件（保存在爱硬盘空间），同时依旧能打印所有日志信息到平台上。

###绑定

在之前的例子里我们已经创建绑定。你可以回顾下代码：

channel.queueBind(queueName, EXCHANGE_NAME, "");

A binding is a relationship between an exchange and a queue. This can be simply read as: the queue is interested in messages from this exchange. 一个绑定是一个交换所和一个队列之间的关系。这个很容易理解为：这个队列是对这 交换机的消息感兴趣。

绑定可以带上额外的路由关键字参数。为了消除对basic_publish参数的迷惑，我们将会将它称之为绑定关键字。以下是我们如何通过一个关键字创建一个绑定：

channel.queueBind(queueName, EXCHANGE_NAME, "black");

这绑定关键字的意义取决于 交换机类型。这fanout 交换机，我们之前使用的那个，仅仅忽略它的值。

###直接交换 我们当前的日志系统将所有消息广播到所有消费者。我们想扩展它，让其允许依据其严格的规则过滤消息。例如我们可能想让一个往硬盘中写日志消息的程序仅仅接收关键的错误，而不是将硬盘空间浪费在警告和信息的日志消息上。 我们使用fanout类型的交换机，那个不会给我们太多的灵活性-它仅仅能胜任没头脑的广播。

我们可以使用direct类型的交换机来替代。一个direct交换机背后的路由算法是简单的-一个消息将会进入那些队列的绑定关键字与消息中路由关键字匹配的队列中。

为了说明那个，考虑接下来结构： 在这个结构里，我们看见了这direct类型的交换机绑定了两个队列。第一个队列装有orange绑定关键字，这第二个有两个绑定，一个是black绑定关键字并且另一个是green关键之。 在这个结构里，发送到交换机里的消息，其中消息中带路由关键字为orange将要路由到队列Q1上，消息中带路由关键字为black或green将路由到队列Q2上。所有其他类型的消息会被丢弃。 ###多种绑定 将一个绑定关键字绑定到货个队列上是十分合法的。在我们例子中使用绑定关键字black将X和Q1绑定在一起。既然那样，这direct类型的交换机与fanout类型相似，同样会广播这消息到所有符合的队列中。一个路由关键位balck的关键字将会被传递到Q1和Q2。 ###发出日志 我们将会为我们的日志系统使用这个模型。使用direct类型的交换机来代替fanout类型，发送消息。由于这路由关键字我们可以严格的记录。接收程序通过这种方式可以严格接收它想接收的。让我们首先关注发布日志。 总之，我们首先需要创建个交换机。

channel.exchangeDeclare(EXCHANGE_NAME, "direct");

我们准备发送一个消息：

channel.basicPublish(EXCHANGE_NAME, severity, null, message.getBytes());

为了简化这个事情，我们保证这severity是inof,warning,error中的一个。 ###订阅 接收消息如先前那样工作，有一个例外，我们会把每一个我们感兴趣的severity创建一个新的绑定。

String queueName = channel.queueDeclare().getQueue();

for(String severity : argv){    
  channel.queueBind(queueName, EXCHANGE_NAME, severity);
}

###把它们放在一起 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();

        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类的代码：

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();

        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 + "'");
        }
    }
}

如平常那样编译（看指南第一部分，编译和类路径的建议）。为了方便，当我们运行实例是，我们现在使用一个环境变量$CP(在windows环境上是%CP%)表示类路径。 如果你想仅保存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

例如，为了发布一个错误日志信息，仅需要键入：

$ java -cp $CP EmitLogDirect error "Run. Run. Or it will explode."
 [x] Sent 'error':'Run. Run. Or it will explode.'

EmitLogDirect.java source和ReceiveLogsDirect.java source的所有源代码。

阅览指南第五部分，查看如何根据一个模式来监听消息。