RabbitMQ入门案例之Direct模式

Dear 丶 2024-03-16 16:18 49阅读 0赞

## 前言 ##

RabbitMQ的Direct模式是一种可以根据指定路由key，Exchang将消息发送到具有该路由key下的Queue下进行存储。也就类似于将数据写进指定数据库表中。这个路由Key可以类比为SQL语句中的：where routeKey = …

> 官方文档地址：[https://www.rabbitmq.com/getstarted.html][https_www.rabbitmq.com_getstarted.html]

## 什么是Direct模式 ##

RabbitMQ中的Direct模式是一种消息传输模式，通常使用Direct Exchange（直连交换机）实现。

在Direct模式中，生产者将消息发送到交换机，并指定消息的Routing Key（路由键）。交换机会将Routing Key与队列绑定进行匹配，如果匹配成功，则将该消息路由到对应的队列中。如果没有匹配成功，该消息将被丢弃或返回给生产者。在Direct模式中，每个消息只能被一个消费者接收。

Direct模式常用于一对一的场景，例如订单管理系统中将订单分配给特定的处理队列。

通过使用Exchange和Routing Key来进行消息传输，Direct模式实现了消息的有选择性地路由，提高了消息传输的效率，减少了系统负载。  
![在这里插入图片描述][2e15ed63f5dd4cecb73df0396bf764d7.png_pic_center]

## 实操 ##

### 实操准备工作 ###

在开始使用代码进行操作前，我们先到管理界面构造一个Direct交换机,如下图：  
![在这里插入图片描述][3a6967e5022e4a4da4d1ab0b55445351.png_pic_center]  
为其绑定Queue，同时设置这个Queue的route key，如下图：  
![在这里插入图片描述][909ca52ec71e45fc8012d2d8d6e7aeed.png_pic_center]  
最终绑定结果：  
![在这里插入图片描述][b78a766a0b9b4940be72757b385119c0.png_pic_center]  
既然交换机和队列已经准备好，接下来就是准备依赖与代码了

<dependency>
    	<groupId>com.rabbitmq</groupId>
    	<artifactId>amqp-client</artifactId>
    	<version>5.10.0</version>
    </dependency>

> 生产者代码

public class Producer {
        
        public static void main(String[] args) {
        
            // 1: 创建连接工厂
            ConnectionFactory connectionFactory = new ConnectionFactory();
            // 2: 设置连接属性
            connectionFactory.setHost("ip地址");
            connectionFactory.setPort(5672);
            connectionFactory.setVirtualHost("/");
            connectionFactory.setUsername("guest");
            connectionFactory.setPassword("guest");
            Connection connection = null;
            Channel channel = null;
            try {
        
                // 3: 从连接工厂中获取连接
                connection = connectionFactory.newConnection("生产者");
                // 4: 从连接中获取通道channel
                channel = connection.createChannel();
                // 6： 准备发送消息的内容
                String message = "宇宙无敌爱学习";
                String  exchangeName = "direct_exchange";
                String routingKey1 = "class";
                String routingKey2 = "student";
                // 7: 发送消息给中间件rabbitmq-server
                // @params1: 交换机exchange
                // @params2: 队列名称/routingkey
                // @params3: 属性配置
                // @params4: 发送消息的内容
                channel.basicPublish(exchangeName, routingKey1, null, message.getBytes());
                channel.basicPublish(exchangeName, routingKey2, null, message.getBytes());
                System.out.println("消息发送成功!");
            } catch (Exception ex) {
        
                ex.printStackTrace();
                System.out.println("发送消息出现异常...");
            } finally {
        
                // 7: 释放连接关闭通道
                if (channel != null && channel.isOpen()) {
        
                    try {
        
                        channel.close();
                    } catch (Exception ex) {
        
                        ex.printStackTrace();
                    }
                }
                if (connection != null) {
        
                    try {
        
                        connection.close();
                    } catch (Exception ex) {
        
                        ex.printStackTrace();
                    }
                }
            }
        }
    }

> 消费者代码

public class Consumer {
        
        private static Runnable runnable = () -> {
        
            // 1: 创建连接工厂
            ConnectionFactory connectionFactory = new ConnectionFactory();
            // 2: 设置连接属性
            connectionFactory.setHost("ip地址");
            connectionFactory.setPort(5672);
            connectionFactory.setVirtualHost("/");
            connectionFactory.setUsername("guest");
            connectionFactory.setPassword("guest");
            //获取队列的名称
            final String queueName = Thread.currentThread().getName();
            Connection connection = null;
            Channel channel = null;
            try {
        
                // 3: 从连接工厂中获取连接
                connection = connectionFactory.newConnection("生产者");
                // 4: 从连接中获取通道channel
                channel = connection.createChannel();
                // 5: 申明队列queue存储消息
                /*
                 *  如果队列不存在，则会创建
                 *  Rabbitmq不允许创建两个相同的队列名称，否则会报错。
                 *
                 *  @params1： queue 队列的名称
                 *  @params2： durable 队列是否持久化
                 *  @params3： exclusive 是否排他，即是否私有的，如果为true,会对当前队列加锁，其他的通道不能访问，并且连接自动关闭
                 *  @params4： autoDelete 是否自动删除，当最后一个消费者断开连接之后是否自动删除消息。
                 *  @params5： arguments 可以设置队列附加参数，设置队列的有效期，消息的最大长度，队列的消息生命周期等等。
                 * */
                // 这里如果queue已经被创建过一次了，可以不需要定义
                //channel.queueDeclare("queue1", false, false, false, null);
                // 6： 定义接受消息的回调
                Channel finalChannel = channel;
                finalChannel.basicConsume(queueName, true, new DeliverCallback() {
        
                    @Override
                    public void handle(String s, Delivery delivery) throws IOException {
        
                        System.out.println(queueName + "：收到消息是：" + new String(delivery.getBody(), "UTF-8"));
                    }
                }, new CancelCallback() {
        
                    @Override
                    public void handle(String s) throws IOException {
        
                    }
                });
                System.out.println(queueName + "：开始接受消息");
                System.in.read();
            } catch (Exception ex) {
        
                ex.printStackTrace();
                System.out.println("发送消息出现异常...");
            } finally {
        
                // 7: 释放连接关闭通道
                if (channel != null && channel.isOpen()) {
        
                    try {
        
                        channel.close();
                    } catch (Exception ex) {
        
                        ex.printStackTrace();
                    }
                }
                if (connection != null && connection.isOpen()) {
        
                    try {
        
                        connection.close();
                    } catch (Exception ex) {
        
                        ex.printStackTrace();
                    }
                }
            }
        };
        public static void main(String[] args) {
        
            // 启动三个线程去执行
            new Thread(runnable, "queue1").start();
            new Thread(runnable, "queue2").start();
            new Thread(runnable, "queue3").start();
        }
    }

在生产者代码中，我们定义了两个route key，如下图：  
![在这里插入图片描述][d1accf6e0d0f40b68917dc92d2543f35.png_pic_center]  
在这两个路由key的驱使下，生产者的消息便只会被放到我们刚刚在direct\_exchange交换机中具有这两个路由key的Queue中，我们来执行代码验证一下。

生产者执行结果  
![在这里插入图片描述][bf861ea97e254bba96559028b4b3b726.png_pic_center]  
管理界面效果  
![在这里插入图片描述][eecc70e162cf4af4a559d7ec74165af4.png_pic_center]  
可以看出，消息就只放进了queue2和queue3中，这是符合我们预期的。  
消费者执行结果，如下:![在这里插入图片描述][1844d45c75ac4c15b86aec9e2f47266d.png_pic_center]  
管理界面效果：  
![在这里插入图片描述][bfd8b3fead4c4001b54eb3eeb82c3f14.png_pic_center]  
可以看出，消息也被成功取出去。

以上便是Direct模式的全部内容，仅个人笔记使用 感谢阅读

[https_www.rabbitmq.com_getstarted.html]: https://www.rabbitmq.com/getstarted.html
[2e15ed63f5dd4cecb73df0396bf764d7.png_pic_center]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/16/63a23144b47c4b0d9e17a27960c3b27f.png
[3a6967e5022e4a4da4d1ab0b55445351.png_pic_center]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/16/7ca6a53ae2c740b899dc833485dc2868.png
[909ca52ec71e45fc8012d2d8d6e7aeed.png_pic_center]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/16/6b20e1c8fa6f4f71bc66a502b6261c2d.png
[b78a766a0b9b4940be72757b385119c0.png_pic_center]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/16/e3bda1a539b04575812eb6f9f8cba379.png
[d1accf6e0d0f40b68917dc92d2543f35.png_pic_center]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/16/1eb970aebd10458a924398e03e84e41a.png
[bf861ea97e254bba96559028b4b3b726.png_pic_center]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/16/ca54e69ba41e498a8cf970ab425714ac.png
[eecc70e162cf4af4a559d7ec74165af4.png_pic_center]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/16/3498e3698d614d7f88bb14698fc1037d.png
[1844d45c75ac4c15b86aec9e2f47266d.png_pic_center]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/16/67a8f7cf4d4d445083df27138e7623f0.png
[bfd8b3fead4c4001b54eb3eeb82c3f14.png_pic_center]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/16/259a9189c4eb49fc9b2f5e378035460a.png