Routing Direct
在Fanout模式中,一条消息,会被所有订阅的队列都消费。但是在某些场景下,我们希望不同的消息被不同的队列消费。这时就要用到Direct类型的Exchange。
- 队列与交换机的绑定,不能是任意绑定了,而是要指定一个
RoutingKey
(路由key) - 消息的发送方在向 Exchange发送消息时,也必须指定消息的
RoutingKey
。 - Exchange不再把消息交给每一个绑定的队列,而是根据消息的
Routing Key
进行判断,只有队列的Routingkey
与消息的Routing key
完全一致,才会接收到消息
- P:生产者,向Exchange发送消息,发送消息时,会指定一个routing key。
- X:Exchange(交换机),接收生产者的消息,然后把消息递交给与routing key完全匹配的队列。
- C1:消费者,其所在队列指定了需要routing key 为 error 的消息。
- C2:消费者,其所在队列指定了需要routing key 为 info、error、warning 的消息。
创建生产者
public class MyProducer {
@Test
public void test() throws Exception {
// 交换机
String exchange = "logs_direct";
// 创建工厂
ConnectionFactory factory = new ConnectionFactory();
factory.setVirtualHost("/");
factory.setHost("xuewei.world");
factory.setUsername("xuewei");
factory.setPassword("123456");
factory.setPort(5672);
// 创建连接和通道
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();
// 声明交换机
channel.exchangeDeclare(exchange, "direct");
for (int i = 0; i < 3; i++) {
// 发布消息
channel.basicPublish(exchange, "DEBUG", null, ("DEBUG LOG -> " + i).getBytes());
channel.basicPublish(exchange, "INFO", null, ("INFO LOG -> " + i).getBytes());
channel.basicPublish(exchange, "WARN", null, ("WARN LOG -> " + i).getBytes());
channel.basicPublish(exchange, "ERROR", null, ("ERROR LOG -> " + i).getBytes());
}
}
}
创建消费者1
public class MyConsumer1 {
public static void main(String[] args) throws Exception {
// 指定交换机
String exchange = "logs_direct";
// 创建工厂
ConnectionFactory factory = new ConnectionFactory();
factory.setVirtualHost("/");
factory.setHost("xuewei.world");
factory.setUsername("xuewei");
factory.setPassword("123456");
factory.setPort(5672);
// 创建连接和通道
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();
// 绑定交换机
channel.exchangeDeclare(exchange, "direct");
// 创建临时队列
String queue = channel.queueDeclare().getQueue();
// 将临时队列绑定exchange
channel.queueBind(queue, exchange, "WARN");
channel.queueBind(queue, exchange, "ERROR");
// 处理消息
channel.basicConsume(queue, true, new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
System.out.println("消费者1: " + new String(body));
// TODO 业务处理
}
});
}
}
创建消费者2
public class MyConsumer2 {
public static void main(String[] args) throws Exception {
// 指定交换机
String exchange = "logs_direct";
// 创建工厂
ConnectionFactory factory = new ConnectionFactory();
factory.setVirtualHost("/");
factory.setHost("xuewei.world");
factory.setUsername("xuewei");
factory.setPassword("123456");
factory.setPort(5672);
// 创建连接和通道
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();
// 绑定交换机
channel.exchangeDeclare(exchange, "direct");
// 创建临时队列
String queue = channel.queueDeclare().getQueue();
// 将临时队列绑定exchange
channel.queueBind(queue, exchange, "DEBUG");
channel.queueBind(queue, exchange, "INFO");
// 处理消息
channel.basicConsume(queue, true, new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
System.out.println("消费者2: " + new String(body));
// TODO 业务处理
}
});
}
}
原文地址:https://blog.csdn.net/weixin_53287520/article/details/134701387
本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。
如若转载,请注明出处:http://www.7code.cn/show_11655.html
如若内容造成侵权/违法违规/事实不符,请联系代码007邮箱:suwngjj01@126.com进行投诉反馈,一经查实,立即删除!
声明:本站所有文章,如无特殊说明或标注,均为本站原创发布。任何个人或组织,在未征得本站同意时,禁止复制、盗用、采集、发布本站内容到任何网站、书籍等各类媒体平台。如若本站内容侵犯了原著者的合法权益,可联系我们进行处理。