RabbitMQ消息队列：Routing 消息路由

咱们构建了一个简单的日志系统。接下来，咱们将丰富它：可以使用不一样的severity来监听不一样等级的log。好比咱们但愿只有error的log才保存到磁盘上。

一、绑定（Bindings）

在上一篇博客中咱们已经使用过绑定。相似下面的代码：
channel.queueBind(queueName, EXCHANGE_NAME, "");
绑定表示转发器与队列之间的关系。咱们也能够简单的认为：队列对该转发器上的消息感兴趣。
绑定能够附带一个额外的参数routingKey。为了与避免basicPublish方法（发布消息的方法）的参数混淆，咱们准备把它称做绑定键（binding key）。下面展现如何使用绑定键（binding key）来建立一个绑定：
channel.queueBind(queueName, EXCHANGE_NAME, "black");
绑定键的意义依赖于转发器的类型。对于fanout类型，忽略此参数。

二、直接转发（Direct exchange）

上一篇的日志系统广播全部的消息给全部的消费者。咱们但愿能够对其扩展，来容许根据日志的严重性进行过滤日志。例如：咱们可能但愿把致命类型的错误写入硬盘，而不把硬盘空间浪费在警告或者消息类型的日志上。
以前咱们使用fanout类型的转发器，可是并无给咱们带来更多的灵活性：仅仅能够愚蠢的转发。
咱们将会使用direct类型的转发器进行替代。direct类型的转发器背后的路由转发算法很简单：消息会被推送至绑定键（binding key）和消息发布附带的选择键（routing key）彻底匹配的队列。
图解：

上图，咱们能够看到direct类型的转发器与两个队列绑定。第一个队列与绑定键orange绑定，第二个队列与转发器间有两个绑定，一个与绑定键black绑定，另外一个与green绑定键绑定。
这样的话，当一个消息附带一个选择键（routing key） orange发布至转发器将会被导向到队列Q1。消息附带一个选择键（routing key）black或者green将会被导向到Q2.全部的其余的消息将会被丢弃。

三、多重绑定（multiple bindings）

使用一个绑定键（binding key）绑定多个队列是彻底合法的。如上图，一个附带选择键（routing key）的消息将会被转发到Q1和Q2。

四、发送日志（Emittinglogs）

咱们准备将这种模式用于咱们的日志系统。咱们将消息发送到direct类型的转发器而不是fanout类型。咱们将把日志的严重性做为选择键（routing key）。这样的话，接收程序能够根据严重性来选择接收。咱们首先关注发送日志的代码：

像之前同样，咱们须要先建立一个转发器：

channel.exchangeDeclare(EXCHANGE_NAME, "direct", true, true, null);

而后咱们准备发送一条消息：

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

为了简化代码，咱们假定‘severity’是‘info’，‘warning’，‘error’中的一个。

五、订阅

接收消息的代码和前面的博客的中相似，只有一点不一样：咱们给咱们所感兴趣的严重性类型的日志建立一个绑定。

StringqueueName = channel.queueDeclare().getQueue();

for(Stringseverity : argv)

{

channel.queueBind(queueName, EXCHANGE_NAME, severity);

}

六、完整的实例

package event;

import java.util.Random;
import java.util.UUID;

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

public class EmitLogDirect
{

 private static final String EXCHANGE_NAME = "ex_logs_direct";
 private static final String[] SEVERITIES = { "info", "warning", "error" };

 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();
  // 声明转发器的类型
  //向全部绑定了相应routing key的队列发送消息
        //若是producer在发布消息时没有consumer在监听，消息将被丢弃
        //若是有多个consumer监听了相同的routing key  则他们都会受到消息
  channel.exchangeDeclare(EXCHANGE_NAME, "direct");

  
       
  //发送6条消息
  for (int i = 0; i < 6; i++)
  {
   String severity = getSeverity();
   String message = severity + "_log :" + UUID.randomUUID().toString();
   // 发布消息至转发器，指定routingkey
   //向server发布一条消息
            //参数1：exchange名字，若为空则使用默认的exchange
            //参数2：routing key
            //参数3：其余的属性
            //参数4：消息体
            //RabbitMQ默认有一个exchange，叫default exchange，它用一个空字符串表示，它是direct exchange类型，
            //任何发往这个exchange的消息都会被路由到routing key的名字对应的队列上，若是没有对应的队列，则消息会被丢弃
   channel.basicPublish(EXCHANGE_NAME, severity, null, message
     .getBytes());
   System.out.println(" [x] Sent '" + message + "'");
  }

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

 /**
  * 随机产生一种日志类型
  *
  * @return
  */
 private static String getSeverity()
 {
  Random random = new Random();
  int ranVal = random.nextInt(3);
  return SEVERITIES[ranVal];
 }
}

接收端：

package event;

import java.util.Random;

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

public class ReceiveLogsDirect
{

 private static final String EXCHANGE_NAME = "ex_logs_direct";
 private static final String[] SEVERITIES = { "info", "warning", "error" };

 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();
  // 声明direct类型转发器
  channel.exchangeDeclare(EXCHANGE_NAME, "direct");

  String queueName = channel.queueDeclare().getQueue();
  String severity = getSeverity();
  System.out.println(severity);
  // 指定binding_key
  channel.queueBind(queueName, EXCHANGE_NAME, severity);
  System.out.println(" [*] Waiting for "+severity+" logs. 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());

   System.out.println(" [x] Received '" + message + "'");
  }
 }

 /**
  * 随机产生一种日志类型
  *
  * @return   */ private static String getSeverity() {  Random random = new Random();  int ranVal = random.nextInt(3);  return SEVERITIES[ranVal]; }}