【RabbitMQ消息中间件】8.路由模式
版权声明：本文为博主原创文章，遵循 CC 4.0 BY-SA 版权协议，转载请附上原文出处链接和本声明。
本文链接：https://blog.csdn.net/u013517797/article/details/79517582
上一篇讲解了RabbitMQ的“订阅模式”，本篇来讲解RabbitMQ的队列的“路由模式”。
其实严格来讲，RabbitMQ只有三种模式：“简单模式”、“work模式”以及“交换机模式”。
对于交换机模式来说，又分三种：“订阅模式”、“路由模式”、“通配符模式”，而他们之间的不同就是交换机类型的不同。
目前在实际开发中我们可能会遇到这种问题，生产者发布到消息队列的信息，消费者不一定全部需要，但是使用订阅模式的话，所有消费者都能够拿到，不满足只拿到自己所需信息的需求。
对于以上的需求，“路由模式”就可以进行实现。
路由模式的图示如下：
还是一个生产者、一个交换机，多个绑定交换机的队列，和多个连接不同队列的消费者。与之前的订阅模式不同的是，这里指定了交换机的类型是“direct”；而消费者将队列绑定到交换机时，指定了一个路由Key(error/info/warning等类型)。
如果生产者发送的消息中的路由key是“error”，那么消费者c1和c2都可以接收到，而如果生产者发送的消息中的路由kay是“info”或者“warning”，则只有消费者c2能够收到。
由此可见。“路由模式”的作用就是可以让消费者有选择性的接收消息。
“路由模式”的实现代码和之前的“订阅模式”的生产者和消费者代码差不多，对于生产者：
package cn.jack.rabbitmq.routing;
import cn.jack.rabbitmq.connection.ConnectionUtil;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
public class Send {
private final static String EXCHANGE_NAME="test_exchange_direct";
public static void main(String[] args) throws Exception {
//获取到连接以及mq通道
Connection connection = ConnectionUtil.getConnection();
//从连接中创建通道
Channel channel = connection.createChannel();
//声明Exchange，指定交换机的类型为direct
channel.exchangeDeclare(EXCHANGE_NAME, "direct");
//消息内容1 info类型
String message = "Hello World!";
channel.basicPublish(EXCHANGE_NAME, "info",null, message.getBytes());
System.out.println("[product] Send '"+ message +"'");
//消息内容2 error类型
message = "Has Error!";
channel.basicPublish(EXCHANGE_NAME, "error",null, message.getBytes());
System.out.println("[product] Send '"+ message +"'");
//消息内容3 warning类型
message = "Has Warning!";
channel.basicPublish(EXCHANGE_NAME, "warning",null, message.getBytes());
System.out.println("[product] Send '"+ message +"'");
//关闭通道和连接
channel.close();
connection.close();
}
}
在上面的代码中，首先生产者绑定交换机的时候，指定类型为“direct”，然后发布消息时指定不同消息的路由key。
然后消费者1的代码：
package cn.jack.rabbitmq.routing;
import cn.jack.rabbitmq.connection.ConnectionUtil;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.QueueingConsumer;
public class Recv1 {
private final static String QUEUE_NAME = "test_queue_direct_1";//队列名称
private final static String EXCHANGE_NAME="test_exchange_direct";//交换机名称
public static void main(String[] argv) throws Exception {
// 获取到连接以及mq通道
Connection connection = ConnectionUtil.getConnection();
Channel channel = connection.createChannel();
// 声明队列
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
//绑定队列到交换机，并指定了两个路由key
channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "info");
channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "error");
// 同一时刻服务器只会发一条消息给消费者
channel.basicQos(1);
// 定义队列的消费者
QueueingConsumer consumer = new QueueingConsumer(channel);
// 监听队列，手动返回完成
channel.basicConsume(QUEUE_NAME, false, consumer);
// 获取消息
while (true) {
QueueingConsumer.Delivery delivery = consumer.nextDelivery();
String message = new String(delivery.getBody());
System.out.println(" [consumer1] Received '" + message + "'");
//休眠10ms
Thread.sleep(10);
// 返回确认状态
channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
}
}
}
对于消费者1，在进行绑定队列到交换机时，设置了两个路由key，也就是它只需要接受路由key为“info”和“error”的消息。
然后消费者2的代码：
package cn.jack.rabbitmq.routing;
import cn.jack.rabbitmq.connection.ConnectionUtil;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.QueueingConsumer;
public class Recv2 {
private final static String QUEUE_NAME = "test_queue_direct_2";//队列名称
private final static String EXCHANGE_NAME="test_exchange_direct";//交换机名称
public static void main(String[] argv) throws Exception {
// 获取到连接以及mq通道
Connection connection = ConnectionUtil.getConnection();
Channel channel = connection.createChannel();
// 声明队列
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
//绑定队列到交换机，并指定了一个路由key
channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "warning");
// 同一时刻服务器只会发一条消息给消费者
channel.basicQos(1);
// 定义队列的消费者
QueueingConsumer consumer = new QueueingConsumer(channel);
// 监听队列，手动返回完成
channel.basicConsume(QUEUE_NAME, false, consumer);
// 获取消息
while (true) {
QueueingConsumer.Delivery delivery = consumer.nextDelivery();
String message = new String(delivery.getBody());
System.out.println(" [consumer2] Received '" + message + "'");
//休眠10ms
Thread.sleep(10);
// 返回确认状态
channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
}
}
}
对于消费者2，在进行绑定队列到交换机时，设置了一个路由key，也就是它只需要接受路由key为“warning”的消息。
下面首先运行两个消费者，然后运行生产者，他们的发送与接收情况如下：
可以看到，生产者一共发送了三种路由key的消息，而消费者按照自己绑定队列时的设置接收到需要的路由key的信息。
原文链接：https://blog.csdn.net/acmman/article/details/79517582