RabbitMQ工作队列模式

原创

会洗碗的CV工程师

发布于 2024-04-26 16:47:33

2630

发布于 2024-04-26 16:47:33

文章被收录于专栏：消息中间件

一、概念

与简单模式相比，工作队列模式(Work Queue)多了一些消费者，该模式也使用direct交换机，应用于处理消息较多的情况。特点如下：

一个队列对应多个消费者。
一条消息只会被一个消费者消费。
消息队列默认采用轮询的方式将消息平均发送给消费者。

RabbitMQ的工作队列模式（Work Queues）也被称为任务队列模式，是一种用于处理分布式任务的工作模式。在这种模式下，多个消费者可以监听同一个队列，但每个消息只能被一个消费者处理。这种模式适用于分布式任务处理场景，多个消费者共享处理一组任务，从而提高系统的并发性能和吞吐量。

工作队列模式的特点如下：

多个消费者监听同一队列：多个消费者可以同时连接到同一个队列，并等待接收消息。
消息的唯一处理：尽管多个消费者监听同一队列，但每个消息只能被一个消费者处理。RabbitMQ通过内部机制确保消息的唯一性，避免重复处理。
动态负载均衡：RabbitMQ会根据消费者的处理能力动态分配消息。如果某个消费者处理速度较快，它会接收到更多的消息；如果某个消费者处理速度较慢或处于空闲状态，它会接收到较少的消息。这种动态负载均衡机制有助于优化资源利用和提高系统性能。

工作队列模式适用于以下场景：

分布式任务处理：当需要将一组任务分配给多个消费者并行处理时，可以使用工作队列模式。例如，在一个大型Web应用中，可能需要将用户上传的图片进行压缩、水印处理等任务分配给多个工作节点并行执行。
高并发场景：在需要处理大量并发请求的场景中，可以使用工作队列模式来提高系统的吞吐量和响应速度。通过将请求分发到多个消费者进行处理，可以充分利用系统资源并减少单个消费者的处理压力。

实现工作队列模式的基本步骤如下：

创建队列：在生产者端，首先需要创建一个队列用于存储待处理的任务消息。
发送消息：生产者将任务消息发送到该队列中。
消费者监听：在消费者端，多个消费者可以同时连接到该队列，并等待接收消息。
消息处理：当消费者接收到消息时，它会执行相应的任务处理逻辑。处理完成后，消费者可以向RabbitMQ发送确认消息，表示该消息已被成功处理。
动态分配：RabbitMQ会根据消费者的处理能力和当前负载情况，动态地将消息分配给各个消费者进行处理。

通过工作队列模式，RabbitMQ可以有效地实现分布式任务处理和负载均衡，提高系统的性能和可靠性。

二、编写生产者

和普通模式一样：

public class Producer {
    public static void main(String[] args) throws IOException, TimeoutException {
        // 1.创建链接工厂
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("192.168.66.100");
        connectionFactory.setPort(5672);
        connectionFactory.setUsername("guest");
        connectionFactory.setPassword("guest");
        connectionFactory.setVirtualHost("/");
        // 2.建立连接
        Connection connection = connectionFactory.newConnection();
        // 3.建立信道
        Channel channel = connection.createChannel();
        /**
         * 4.创建队列，如果队列已存在，则使用该队列
         * 参数1：队列名
         * 参数2：是否持久化，true表示MQ重启后队列还在。
         * 参数3：是否私有化，false表示所有消费者都可以访问，true表示只有第一次拥有它的消费者才能访问
         * 参数4：是否自动删除，true表示不再使用队列时自动删除队列
         * 参数5：其他额外参数
         */
        channel.queueDeclare("work_queue",true,false,false,null);
        // 5.发送大量消息
        for (int i=1; i<=100; i++){
            /**
             * 参数1：交换机名，""表示默认交换机
             * 参数2：路由键，简单模式就是队列名
             * 参数3：表示该消息为持久化消息，即除了保存到内存还会保存到磁盘中
             * 参数4：要传递的消息字节数组
             */
            channel.basicPublish("","work_queue", MessageProperties.PERSISTENT_TEXT_PLAIN,("你好这是第"+i+"消息").getBytes());
        }

        // 6.关闭信道和连接
        channel.close();
        connection.close();
        System.out.println("--- 发送成功 ---");
    }
}

运行成功如下图：

三、编写消费者

这里我们可以编写三个消费者，查看他们获取的消息

public class Producer {
    public static void main(String[] args) throws IOException, TimeoutException {
        // 1.创建链接工厂
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("192.168.66.100");
        connectionFactory.setPort(5672);
        connectionFactory.setUsername("guest");
        connectionFactory.setPassword("guest");
        connectionFactory.setVirtualHost("/");
        // 2.建立连接
        Connection connection = connectionFactory.newConnection();
        // 3.建立信道
        Channel channel = connection.createChannel();
        /**
         * 4.创建队列，如果队列已存在，则使用该队列
         * 参数1：队列名
         * 参数2：是否持久化，true表示MQ重启后队列还在。
         * 参数3：是否私有化，false表示所有消费者都可以访问，true表示只有第一次拥有它的消费者才能访问
         * 参数4：是否自动删除，true表示不再使用队列时自动删除队列
         * 参数5：其他额外参数
         */
        channel.queueDeclare("work_queue",true,false,false,null);
        // 5.发送大量消息
        for (int i=1; i<=100; i++){
            /**
             * 参数1：交换机名，""表示默认交换机
             * 参数2：路由键，简单模式就是队列名
             * 参数3：表示该消息为持久化消息，即除了保存到内存还会保存到磁盘中
             * 参数4：要传递的消息字节数组
             */
            channel.basicPublish("","work_queue", MessageProperties.PERSISTENT_TEXT_PLAIN,("你好这是第"+i+"消息").getBytes());
        }

        // 6.关闭信道和连接
        channel.close();
        connection.close();
        System.out.println("--- 发送成功 ---");
    }
}

运行后如下图：

消费者1如下图：