RabbitMQ系列学习笔记（九）--路由模式

本文参考
尚硅谷RabbitMQ教程丨快速掌握MQ消息中间件rabbitmq
RabbitMQ 详解
Centos7环境安装Erlang、RabbitMQ详细过程(配图)

一、路由模式原理

使用发布订阅模式时，所有消息都会发送到绑定的队列中，但很多时候，不是所有消息都要无差别的发布到所有队列中。比如，我们希望将日志消息写入磁盘的程序仅接收严重错误(errros)，而不存储那些警告(warning)或信息(info)日志消息以避免浪费磁盘空间。Fanout这种交换类型并不能给我们带来很大的灵活性-它只能进行无意识的广播，此时就需要使用路由模式 (Routing)完成这一需求。其特点如下：

• 每个队列绑定路由关键字 RoutingKey
• 生产者将带有 RoutingKey 的消息发送给交换机，交换机根据 RoutingKey 转发到指定队列。
• 路由模式使用 direct 交换机。

回顾bindings的概念，绑定是交换机和队列之间的桥梁关系。也可以这么理解：队列只对它绑定的交换机的消息感兴趣。绑定用参数：routingKey来表示也可称该参数为binding key，创建绑定我们用代码：channel.queueBind(queueName, EXCHANGE_NAME, "routingKey");
在上图实例中，可以看到交换机 X 绑定了两个队列，绑定类型是direct。队列Q1绑定键为orange，队列Q2绑定键有两个：一个绑定键为black，另一个绑定键为green。这说明即使是同一对交换机和队列也可以有多个routingKey。
在这种绑定情况下，生产者发布消息到exchange上，绑定键为orange的消息会被发布到队列Q1。绑定键为black和green和的消息会被发布到队列Q2，其他消息类型的消息将被丢弃。

二、多重绑定

如果exchange的绑定类型是direct，但是它绑定的多个队列的key如果都相同，在这种情况下虽然绑定类型是direct但是它表现的就和fanout有点类似了，就跟广播差不多，如上图所示。

三、路由模式实战

根据以上对应关系进行代码编写，其中包括两个队列：disk和console，一个生产者，两个消费者，一个direct类型的交换，与disk队列通过error进行绑定，与console队列通过info和warning两个routingkey进行绑定。最终进行测试，查看结果。

1、消费者代码

同样的，代码整体逻辑与之前的案例没什么不同，只是交换机类型变更，以及需要指定routingkey参数。
消费者01代码如下：

/*** Description: 路由模式消费者01*/
public class ReceiveLogsDirect01 {//设置要创建的交换机的名称private static final String EXCHANGE_NAME = "direct_logs";public static void main(String[] args) throws Exception {Channel channel = RabbitMqUtil.getChannel();//创建fanout交换机/*** 参数1：交换机名* 参数2：交换机类型，本次设置为direct* 参数3：交换机是否持久化*/channel.exchangeDeclare(EXCHANGE_NAME, "direct", false);channel.queueDeclare("disk", false, false, false, null);//将交换机与队列进行绑定（binding）/*** 参数1：队列名* 参数2：交换机名* 参数3：路由关键字，发布订阅模式写""空串即可*/channel.queueBind("disk", EXCHANGE_NAME, "error");//接收消息channel.basicConsume("disk", true, new DefaultConsumer(channel) {@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {String message = new String(body, "UTF-8");System.out.println("ReceiveLogsDirect01控制台打印接收到的消息: " + message);}});}
}

消费者02代码如下：

/*** Description: 路由模式消费者02*/
public class ReceiveLogsDirect02 {//设置要创建的交换机的名称private static final String EXCHANGE_NAME = "direct_logs";public static void main(String[] args) throws Exception {Channel channel = RabbitMqUtil.getChannel();channel.exchangeDeclare(EXCHANGE_NAME, "direct", false);channel.queueDeclare("console", false, false, false, null);//多重绑定channel.queueBind("console", EXCHANGE_NAME, "info");channel.queueBind("console", EXCHANGE_NAME, "warning");//接收消息channel.basicConsume("console", true, new DefaultConsumer(channel) {@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {String message = new String(body, "UTF-8");System.out.println("ReceiveLogsDirect02控制台打印接收到的消息: " + message);}});}
}

2、生产者代码

/*** Description: 路由模式生产者*/
public class EmitLogDirect {//交换机名称private static final String EXCHANGE_NAME = "direct_logs";public static void main(String[] args) throws Exception {Channel channel = RabbitMqUtil.getChannel();//准备3条消息String message1 = "这是一条info信息";String message2 = "这是一条warning信息";String message3 = "这是一条error信息";channel.basicPublish(EXCHANGE_NAME, "info", null, message1.getBytes());channel.basicPublish(EXCHANGE_NAME, "warning", null, message2.getBytes());channel.basicPublish(EXCHANGE_NAME, "error", null, message3.getBytes());//关闭资源channel.close();}
}

3、运行结果分析

先将两个消费者运行，因为它们负责声明交换机创建队列以及绑定关系，再启动生产者发送消息，此时会看到message1和message2发给了ReceiveLogsDirect02，而message3发送给了ReceiveLogsDirect01
在此过程中，生产者在发送消息时指定了EXCHANGE_NAME，无论是什么消息都先发送给direct类型的交换机，它的名字设置为了direct_logs，然后交换机会根据routingkey的路由规则决定该消息转发给哪个队列，以及是否要丢弃。假如此时我们发送一个routingkey为debug的消息，交换机由于找不到转发目标会将该消息丢弃。