rabbitMQ_routing(四)

路由

本次我们将通过路由将信息发送到指定的队列中，将消息发送给指定的队列需要在转发器和队列之间建立一个routeKey

绑定

在以前的例子中，我们已经创建了绑定。你可能会记得如下代码：

channel.queueBind（queueName，EXCHANGE_NAME，“”）;
以上代码是针对于fanout转发器，这种类型的转发器会广播消息到所以与它绑定了的队列，会忽略第三个参数
本例将使用direct类型的转发器，这种转发器需要指定routeKey,告诉这个转发器，消息应当转发给哪个队列

channel.queueBind（queueName，EXCHANGE_NAME，“black”）;

以上代码中的第三个参数black就是routekey。

  routeKey的含义取决于转发类型。第三个参数是routeKey,如果转发器类型是fanout，那么这个routeKey会被忽略。

direct转发器

直接转发器的路由算法很简单 - 消息只会传递给routeKey匹配的队列。如下图：

 

上图中，我们可以看到直接转发器X与两个绑定的队列。第一个队列与routeKey橙色绑定，第二个队列有两个绑定，一个routeKey为黑色，另一个routeKey为绿色。

生产者发送消息给orange这个routeKey，那么消息会被转发到Q1这个队列，如果routeKey是black和green，那么就会转发到Q2这个队列。

多重绑定

使用相同的routeKey绑定多个队列是完全合法的。在我们的示例中，我们可以在X和Q1之间添加routeKey black。在这种情况下，direct转发器将表现得像fanout转发器，并将消息广播到所有匹配的队列。具有routeKey为black的消息将传送到 Q1和Q2。

提交日志

本次实例将消息发送到direct转发器，转发器根据routeKey决定消息应当发往哪个队列，我们将以日志的严重程度(error, warning, info)作为routeKey。

我们需要定义一个直接转发器：

channel.exchangeDeclare（EXCHANGE_NAME，“direct”）;

我们准备发送消息：

channel.basicPublish（EXCHANGE_NAME，severity，null，message.getBytes（））;

severity为error，info，warning之一。

订阅

   创建一个排他，自动删除，随机命名，非持久的队列：

String queueName = channel.queueDeclare（）.getQueue（）;

 使用severity绑定队列
for（String severity：argv）{
  channel.queueBind（queueName，EXCHANGE_NAME，severity）;
}

下图为绑定示意图：

EmitLogDirect.java类的代码：

package com.rabbitMQ;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

public class EmitLogDirect {

    private static final String EXCHANGE_NAME = "direct_logs";

    public static void main(String[] argv) throws java.io.IOException, Exception {

        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);
        // 第二参数为routeKey，指定消息应当发给哪个队列
        channel.basicPublish(EXCHANGE_NAME, severity, null, message.getBytes());
        System.out.println(" [x] Sent '" + severity + "':'" + message + "'");

        channel.close();
        connection.close();
    }

    private static String getSeverity(String[] argv) {

        if (argv.length == 0) {

            return "info";
        }
        String str = argv[0];

        String routeKey = str.replaceAll("=.*", "");

        return routeKey;
    }

    private static String getMessage(String[] argv) {
        if (argv.length == 0) {

            return "mistake happen!!!";
        }
        String str = argv[0];
        String message = str.replaceAll(".*=", "");
        return message;
    }

}

ReceiveLogsDirect.java的代码：

package com.rabbitMQ;

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

    //给转发器和队列之间建立routeKey,传入info，warning，error等级
    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);
  }
}

启动ReceiveLogsDirect.java，给它传error，warning，info参数

再次启动ReceiveLogsDirect.java，给它传warning，info参数

启动EmitLogDirect.java，给它传error=some error happend

那么只有第一个ReceiveLogsDirect.java才能接收到消息，

再次启动EmitLogDirect.java，给它传info=Send message

那么两个ReceiveLogsDirect.java都会接收到消息。