RabbitMQ消息模型之Routing-Direct

简介: RabbitMQ消息模型之Routing-Direct

Routing Direct

在Fanout模式中,一条消息,会被所有订阅的队列都消费。但是在某些场景下,我们希望不同的消息被不同的队列消费。这时就要用到Direct类型的Exchange。

在Direct模型下:

  • 队列与交换机的绑定,不能是任意绑定了,而是要指定一个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 业务处理
            }
        });
    }
}


相关实践学习
RocketMQ一站式入门使用
从源码编译、部署broker、部署namesrv,使用java客户端首发消息等一站式入门RocketMQ。
消息队列 MNS 入门课程
1、消息队列MNS简介 本节课介绍消息队列的MNS的基础概念 2、消息队列MNS特性 本节课介绍消息队列的MNS的主要特性 3、MNS的最佳实践及场景应用 本节课介绍消息队列的MNS的最佳实践及场景应用案例 4、手把手系列:消息队列MNS实操讲 本节课介绍消息队列的MNS的实际操作演示 5、动手实验:基于MNS,0基础轻松构建 Web Client 本节课带您一起基于MNS,0基础轻松构建 Web Client
相关文章
|
3天前
|
消息中间件
RabbitMQ消息模型之发布订阅Publish-Subscribe
RabbitMQ消息模型之发布订阅Publish-Subscribe
19 0
RabbitMQ消息模型之发布订阅Publish-Subscribe
|
3天前
|
消息中间件 JSON 缓存
RabbitMQ快速学习之WorkQueues模型、三种交换机、消息转换器(SpringBoot整合)
RabbitMQ快速学习之WorkQueues模型、三种交换机、消息转换器(SpringBoot整合)
|
3天前
|
消息中间件 JSON 缓存
RabbitMQ快速学习之WorkQueues模型、三种交换机、消息转换器(基于SpringBoot)
RabbitMQ快速学习之WorkQueues模型、三种交换机、消息转换器(基于SpringBoot)
|
3天前
|
消息中间件 Java Kafka
RabbitMQ安装和5种不同的消息模型(BasicQueue,WorkQueue,Fanout Exchange,Direct Exchange,Topic Exchange)与SpringAMQP
RabbitMQ安装和5种不同的消息模型(BasicQueue,WorkQueue,Fanout Exchange,Direct Exchange,Topic Exchange)与SpringAMQP
|
10月前
|
消息中间件 算法 Java
RabbitMQ灵活运用,怎么理解五种消息模型
RabbitMQ灵活运用,怎么理解五种消息模型
84 0
|
消息中间件 存储 负载均衡
理解RabbitMQ中的AMQP-0-9-1模型
之前有个打算在学习RabbitMQ之前,把AMQP详细阅读一次,挑出里面的重点内容。后来找了下RabbitMQ的官方文档,发现了有一篇文档专门介绍了RabbitMQ中实现的AMQP模型部分,于是直接基于此文档和个人理解写下这篇文章。
158 0
理解RabbitMQ中的AMQP-0-9-1模型
|
消息中间件 Java Spring
【RabbitMQ】Basic Queue 简单队列模型与WorkQueue
【RabbitMQ】Basic Queue 简单队列模型与WorkQueue
79 0
【RabbitMQ】Basic Queue 简单队列模型与WorkQueue
|
消息中间件 Java Kafka
RabbitMQ安装以及消息模型使用攻略
🍅程序员小王的博客:程序员小王的博客 🍅 欢迎点赞 👍 收藏 ⭐留言 📝 🍅 如有编辑错误联系作者,如果有比较好的文章欢迎分享给我,我会取其精华去其糟粕 🍅java自学的学习路线:java自学的学习路线
203 0
RabbitMQ安装以及消息模型使用攻略
|
消息中间件 Java 中间件
【Alibaba中间件技术系列】「RocketMQ技术专题」带你一起去探索RocketMQ服务架构的线程模型分析
【Alibaba中间件技术系列】「RocketMQ技术专题」带你一起去探索RocketMQ服务架构的线程模型分析
239 0
【Alibaba中间件技术系列】「RocketMQ技术专题」带你一起去探索RocketMQ服务架构的线程模型分析
|
存储 消息中间件 负载均衡
终于弄明白 RocketMQ 存储模型
RocketMQ 优异的性能表现,必然绕不开其优秀的存储模型 。 这篇文章,笔者按照自己的理解 , 尝试分析 RocketMQ 的存储模型,希望对大家有所启发。