RabbitMQ路由模式（direct）

1、什么是路由模式（direct）

　　路由模式是在使用交换机的同时，生产者指定路由发送数据，消费者绑定路由接受数据。与发布/订阅模式不同的是，发布/订阅模式只要是绑定了交换机的队列都会收到生产者向交换机推送过来的数据。而路由模式下加了一个路由设置，生产者向交换机发送数据时，会声明发送给交换机下的那个路由，并且只有当消费者的队列绑定了交换机并且声明了路由，才会收到数据。下图取自于官方网站（RabbitMQ）的路由模式的图例

P：消息的生产者

X：交换机

红色：队列

C1，C2：消息消费者

error，info，warning：路由

　　举个日志处理例子：系统需要针对日志做分析，首先所有的日志级别的日志都需要保存，其次error日志级别的日志需要单独做处理。这时就可以使用路由模式来处理了，声明交换机使用路由模式，每个日志级别的日志对应一个路由（error，info，warning）。声明一个保存日志队列用于接受所有日志，绑定交换机并绑定所有路由。声明第二个队列用于处理error级别日志，绑定交换机且只绑定error路由。以下是代码讲解。（先运行两个消费者，在运行生产者。如果没有提前将队列绑定到交换机，那么直接运行生产者的话，消息是不会发到任何队列里的）

2、生产者（Send）代码

public class Send
{//交换机名称private final static String EXCHANGE_NAME       = "test_exchange_direct";//路由名称warningprivate final static String ROUTING_KEY_WARNING = "warning";//路由名称infoprivate final static String ROUTING_KEY_INFO    = "info";//路由名称errorprivate final static String ROUTING_KEY_ERROR   = "error";public static void main(String[] args){try{//获取连接Connection connection = ConnectionUtil.getConnection();//从连接中获取一个通道Channel channel = connection.createChannel();//声明交换机channel.exchangeDeclare(EXCHANGE_NAME, "direct");String message = "this is warning log";//发送消息(warning级别日志)channel.basicPublish(EXCHANGE_NAME, ROUTING_KEY_WARNING, null, message.getBytes("utf-8"));System.out.println("[send]:" + message);//发送消息(info级别日志)message = "this is info log";channel.basicPublish(EXCHANGE_NAME, ROUTING_KEY_INFO, null, message.getBytes("utf-8"));System.out.println("[send]:" + message);//发送消息(error级别日志)message = "this is error log";channel.basicPublish(EXCHANGE_NAME, ROUTING_KEY_ERROR, null, message.getBytes("utf-8"));System.out.println("[send]:" + message);channel.close();connection.close();}catch (IOException | TimeoutException e){e.printStackTrace();}}
}运行结果：

[send]:this is warning log
[send]:this is info log
[send]:this is error log

3、消费者1（ReceiveAllLog）

public class ReceiveAllLog
{//交换机名称private final static String EXCHANGE_NAME       = "test_exchange_direct";//路由名称warningprivate final static String ROUTING_KEY_WARNING = "warning";//路由名称infoprivate final static String ROUTING_KEY_INFO    = "info";//路由名称errorprivate final static String ROUTING_KEY_ERROR   = "error";//队列名称private static final String QUEUE_NAME          = "test_queue_save_all_log";public static void main(String[] args){try{//获取连接Connection connection = ConnectionUtil.getConnection();//从连接中获取一个通道final Channel channel = connection.createChannel();//声明交换机channel.exchangeDeclare(EXCHANGE_NAME, "direct");//声明队列channel.queueDeclare(QUEUE_NAME, false, false, false, null);//将队列绑定到交换机（指定路由info,error,warning）channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, ROUTING_KEY_INFO);channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, ROUTING_KEY_ERROR);channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, ROUTING_KEY_WARNING);//保证一次只分发一个  int prefetchCount = 1;channel.basicQos(prefetchCount);//定义消费者DefaultConsumer consumer = new DefaultConsumer(channel){//当消息到达时执行回调方法@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties,byte[] body) throws IOException{String message = new String(body, "utf-8");System.out.println("[test_queue_save_all_log] Receive message：" + message);try{//消费者休息2s处理业务Thread.sleep(1000);}catch (InterruptedException e){e.printStackTrace();}finally{//手动应答channel.basicAck(envelope.getDeliveryTag(), false);}}};//设置手动应答boolean autoAck = false;//监听队列channel.basicConsume(QUEUE_NAME, autoAck, consumer);}catch (IOException e){e.printStackTrace();}}
}运行结果：

[test_queue_save_all_log] Receive message：this is warning log
[test_queue_save_all_log] Receive message：this is info log
[test_queue_save_all_log] Receive message：this is error log

4、消费者2（ReceiveErrorLog）

public class ReceiveErrorLog
{//交换机名称private final static String EXCHANGE_NAME     = "test_exchange_direct";//路由名称errorprivate final static String ROUTING_KEY_ERROR = "error";//队列名称private static final String QUEUE_NAME        = "test_queue_handel_error";public static void main(String[] args){try{//获取连接Connection connection = ConnectionUtil.getConnection();//从连接中获取一个通道final Channel channel = connection.createChannel();//声明队列channel.queueDeclare(QUEUE_NAME, false, false, false, null);//将队列绑定到交换机（指定路由error）channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, ROUTING_KEY_ERROR);//保证一次只分发一个  int prefetchCount = 1;channel.basicQos(prefetchCount);//定义消费者DefaultConsumer consumer = new DefaultConsumer(channel){//当消息到达时执行回调方法@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties,byte[] body) throws IOException{String message = new String(body, "utf-8");System.out.println("[test_queue_handel_error] Receive message：" + message);try{//消费者休息2s处理业务Thread.sleep(1000);}catch (InterruptedException e){e.printStackTrace();}finally{//手动应答channel.basicAck(envelope.getDeliveryTag(), false);}}};//设置手动应答boolean autoAck = false;//监听队列channel.basicConsume(QUEUE_NAME, autoAck, consumer);}catch (IOException e){e.printStackTrace();}}
}运行结果：
[test_queue_handel_error] Receive message：this is error log

总结：

　　1.两个队列消费者设置的路由不一样，接收到的消息就不一样。路由模式下，决定消息向队列推送的主要取决于路由，而不是交换机了。

　　2.该模式必须设置交换机，且声明路由模式：channel.exchangeDeclare(EXCHANGE_NAME, "direct");

RabbitMQ路由模式（direct）相关推荐

路由模式 - direct
2019独角兽企业重金招聘Python工程师标准>>> package com.shi.rout;import java.io.IOException; import java.ut ...
RabbitMQ的Routing 路由模式(Direct)
RabbitMQ的Routing 路由模式模式说明: 队列与交换机的绑定,不能是任意绑定了,而是要指定一个 RoutingKey(路由key) 消息的发送方在向 Exchange 发送消息时,也必须 ...
RabbitMQ路由模式
路由模式一个生产者,发送消息每个消费者,都有一个独立的队列消息发送到交换机,交换机发送到每个队列根据key,是否相等,来接收消息 Send 生产者 package cn.itcast.rabb ...
【夏目鬼鬼分享】RabbitMQ路由模式
路由模式路由模式是可以根据路由键选择性给多个消费者发送消息的模式,它包含一个生产者.两个消费者.两个队列和一个交换机.两个消费者同时绑定到不同的队列上去,两个队列通过路由键绑定到交换机上去,生产者发 ...
RabbitMQ——路由模式
下面我们将要实现这个模型,所有的代码将都是以这个模型为基础: direct: 首先,我们设置的routingKey是 error,那么按照路由规则,我们最终将向这两个队列发送消息: 生产者: publ ...
（需求实战_进阶_03）SSM集成RabbitMQ 路由模式关键代码讲解、开发、测试
接上一篇:(企业内部需求实战_进阶_02)SSM集成RabbitMQ 关键代码讲解.开发.测试 https://gblfy.blog.csdn.net/article/details/10421403 ...
【转】RabbitMQ六种队列模式-4.路由模式
前言 RabbitMQ六种队列模式-简单队列 RabbitMQ六种队列模式-工作队列 RabbitMQ六种队列模式-发布订阅 RabbitMQ六种队列模式-路由模式 [本文] RabbitMQ六种队列 ...
RabbitMQ(六) Routing路由模式
概述所谓RabbitMq中路由模式(Routing)为我们在将发送消息队列以及接收消息队列(queue)绑定到交换机(exchange)时指定了一个RoutingKey.然后我们在通过连接信道向交换 ...
【RabbitMQ】基础四：路由模式（Routing）
[RabbitMQ]基础四:路由模式(Routing) 1. 路由模式说明 2. 代码示例 2.1 生产者 2.2 消费者1 2.3 消费者2 2.4 测试 3. 总结 1. 路由模式说明路由模式特 ...

RabbitMQ路由模式（direct）

RabbitMQ路由模式（direct）相关推荐

最新文章

热门文章