RabbitMQ入门学习——Routing(路由)

 

http://www.rabbitmq.com/tutorials/tutorial-four-java.html

                
 

在上一章节中，我们建立了一个简单的日志系统。我们可以将消息广播给许多的接收者

本章节，我们打算添加一个新的特性：我们打算只订阅消息的一部分。比如，我们只将严重的错误信息保存到文件中，同时将所有的信息打印到控制台

1：Bindings(绑定)

在前面的章节中，我们已经创建了绑定。也许你还记得如下代码：channel.queueBind(queueName, EXCHANGE_NAME, "");

绑定是交换区和队列之间的关系。可以简单理解为：队列对交换区中得消息感兴趣。绑定可以通过一个额外的参数routingKey。为了避免和basic_publish参数混淆，我们叫它binding key。我们可以通过下面的方式使用指定的key创建绑定：

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

绑定的意义需要依赖于交换区的类型。前面使用的fanout类型的交换区会忽略routingkey的值。

 

2：Direct exchange(直接交换)

     我们前面的日志程序将所有的消息广播给所有的消费者。我们希望扩展该功能，依据消息的严重程度来过滤消息。例如：我们只希望将严重的消息保存到硬盘上，而不浪费硬盘空间来保存警告和一般的日志信息。

     我们目前使用的fanout类型的交换区不能给我们如此灵活的功能，它只能盲目的广播消息。我们将使用direct类型的交换区代替它。Direct背后的路由算法非常简单，它会将发往队列消息的binding key和routing key进行匹配。为了说明清楚，考虑如下步骤(结合上面图片进行说明)： 

     首先，我们看到有两个队列绑定到direct交换区X上。第一个队列通过绑定key“orange”，第二个绑定了两个key：“black”和“green”。

     在这种情况下，一条通过路由key“orange”发布到交换区上的消息，将会被路由到队列Q1。使用路由key“black”和“green”的消息将进入队列Q2中。所有其他的消息将会被丢弃。

3：Multiple bindings(多个绑定)

使用同一个名字绑定多个队列也是完全合法的。在我们的例子中，我们可以使用绑定key“black”来绑定X和Q1。在这种情况下，direct类型的交换区和fanout类似，会广播消息给所有匹配的队列。拥有路由key“black”的消息，将会被同时传递给Q1和Q2

4：Emitting logs(发送日志)

我们会将这种模式应用到我们的日志程序上。我们将消息发送到一个direct交换区上，而不是之前的fanout交换区。通过这种方式，接收程序可以按照严重程度，选择它需要的消息。让我们先看看发送日志的代码。

像之前一样，我们需要先创建一个交换区

channel.exchangeDeclare(EXCHANGE_NAME, "direct");

然后我们准备发送消息：

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

我们假设‘severity’取值为‘info’，‘warning’，‘error’

 

5：Subscribing(订阅)

接收消息的程序如果想之前那样运行，会抛出一个异常。我们会为每一个我们感兴趣的severity创建一个绑定。

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

 

for(String severity : argv){    

  channel.queueBind(queueName, EXCHANGE_NAME, severity);

}

5：Putting it all together

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

    }

    //..

}

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

        }

    }

}