RabbitMQ六种通信模式介绍——工作模式（Work queues）

╰+攻爆jí腚メ 2022-12-27 02:20 206阅读 0赞

### 文章目录 ###

*   *  一。简介
     *  二。代码实现
     *  三。程序运行效果
     *  四。模式总结

--------------------

## 一。简介 ##

本篇博客所讲的为RabbitMQ六种通信模式之一的工作模式，官网给出的图如下所示：

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDAwOTQ0Nw_size_16_color_FFFFFF_t_70]  
**Work Queues**与入门程序的简单模式相比，多了一个或一些消费端，多个消费端共同消费同一个队列中的消息。

**应用场景：对于任务过重或任务较多情况使用工作队列可以提高任务处理的速度。**

## 二。代码实现 ##

1.连接工具类：

import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.ConnectionFactory;
    
    /** * 创建连接工具类 */
    public class ConnectionUtil { 
    
        public static Connection getConnection() throws Exception { 
            //创建连接工厂
            ConnectionFactory connectionFactory = new ConnectionFactory();
            //主机地址;默认为 localhost
            connectionFactory.setHost("localhost");
            //连接端口;默认为 5672
            connectionFactory.setPort(5672);
            //虚拟主机名称;默认为 /
            connectionFactory.setVirtualHost("hello");
            //连接用户名；默认为guest
            connectionFactory.setUsername("guest");
            //连接密码；默认为guest
            connectionFactory.setPassword("guest");
    
            //创建连接
            return connectionFactory.newConnection();
        }
    
    }

2.消息生产者：

import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    
    /** * 消息生产者 */
    public class Producer { 
        static final String QUEUE_NAME = "rabbitmq_queue";
    
        public static void main(String[] args) throws Exception { 
            //创建连接
            Connection connection = ConnectionUtil.getConnection();
    
            // 创建频道
            Channel channel = connection.createChannel();
    
            // 声明（创建）队列
            /** * 参数1：队列名称 * 参数2：是否定义持久化队列（重启后是否能够再次看见未销毁的队列） * 参数3：是否独占本次连接（是否一个channel占用一个队列） * 参数4：是否在不使用的时候自动删除队列 * 参数5：队列其它参数 */
            channel.queueDeclare(QUEUE_NAME, true, false, false, null);
    
            // 要发送的信息
            for (int i = 1; i <= 30; i++) { 
                String message = "今天天气真好！"+i;
                /** * 参数1：交换机名称，如果没有指定则使用默认Default Exchage * 参数2：路由key,简单模式可以传递队列名称 * 参数3：消息其它属性 * 参数4：消息内容 */
                channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
                System.out.println("已发送消息：" + message);
            }
    
            // 关闭资源
            channel.close();
            connection.close();
        }
    
    }

3.消息消费者1：

import com.rabbitmq.client.*;
    import java.io.IOException;
    
    /** * 消息消费者1 */
    public class Consumer { 
    
        public static void main(String[] args) throws Exception { 
            //创建连接
            Connection connection = ConnectionUtil.getConnection();
    
            // 创建频道
            Channel channel = connection.createChannel();
    
            // 声明（创建）队列
            /** * 参数1：队列名称（消费者与提供者传输的队列名称需要一致） * 参数2：是否定义持久化队列（重启后是否能够再次看见未销毁的队列） * 参数3：是否独占本次连接（是否一个channel占用一个队列） * 参数4：是否在不使用的时候自动删除队列 * 参数5：队列其它参数 */
            channel.queueDeclare(Producer.QUEUE_NAME, true, false, false, null);
            //一次只能接收并处理一个消息
            channel.basicQos(1);
    
            //创建消费者；并设置消息处理
            DefaultConsumer consumer = new DefaultConsumer(channel){ 
                @Override
                /** * consumerTag 消息者标签，在channel.basicConsume时候可以指定 * envelope 消息包的内容，可从中获取消息id，消息routingkey，交换机，消息和重传标志(收到消息失败后是否需要重新发送) * properties 属性信息 * body 消息 */
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { 
                    //路由key
                    System.out.println("路由key为：" + envelope.getRoutingKey());
                    //交换机
                    System.out.println("交换机为：" + envelope.getExchange());
                    //消息id
                    System.out.println("消息id为：" + envelope.getDeliveryTag());
                    //收到的消息
                    System.out.println("消费者1接收到的消息为：" + new String(body, "utf-8"));
                    System.out.println("__________________________________________________________");
                }
            };
            //监听消息
            /** * 参数1：队列名称 * 参数2：是否自动确认，设置为true为表示消息接收到自动向mq回复接收到了，mq接收到回复会删除消息，设置为false则需要手动确认 * 参数3：消息接收到后回调 */
            channel.basicConsume(Producer.QUEUE_NAME, true, consumer);
    
            //不关闭资源，应该一直监听消息
            //channel.close();
            //connection.close();
        }
    }

4.消息消费者2：

import com.rabbitmq.client.*;
    import java.io.IOException;
    
    /** * 消息消费者2 */
    public class Consumer1 { 
    
        public static void main(String[] args) throws Exception { 
            //创建连接
            Connection connection = ConnectionUtil.getConnection();
    
            // 创建频道
            Channel channel = connection.createChannel();
    
            // 声明（创建）队列
            /** * 参数1：队列名称（消费者与提供者传输的队列名称需要一致） * 参数2：是否定义持久化队列（重启后是否能够再次看见未销毁的队列） * 参数3：是否独占本次连接（是否一个channel占用一个队列） * 参数4：是否在不使用的时候自动删除队列 * 参数5：队列其它参数 */
            channel.queueDeclare(Producer.QUEUE_NAME, true, false, false, null);
            //一次只能接收并处理一个消息
            channel.basicQos(1);
    
            //创建消费者；并设置消息处理
            DefaultConsumer consumer = new DefaultConsumer(channel){ 
                @Override
                /** * consumerTag 消息者标签，在channel.basicConsume时候可以指定 * envelope 消息包的内容，可从中获取消息id，消息routingkey，交换机，消息和重传标志(收到消息失败后是否需要重新发送) * properties 属性信息 * body 消息 */
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { 
                    //路由key
                    System.out.println("路由key为：" + envelope.getRoutingKey());
                    //交换机
                    System.out.println("交换机为：" + envelope.getExchange());
                    //消息id
                    System.out.println("消息id为：" + envelope.getDeliveryTag());
                    //收到的消息
                    System.out.println("消费者2接收到的消息为：" + new String(body, "utf-8"));
                    System.out.println("__________________________________________________________");
                }
            };
            //监听消息
            /** * 参数1：队列名称 * 参数2：是否自动确认，设置为true为表示消息接收到自动向mq回复接收到了，mq接收到回复会删除消息，设置为false则需要手动确认 * 参数3：消息接收到后回调 */
            channel.basicConsume(Producer.QUEUE_NAME, true, consumer);
    
            //不关闭资源，应该一直监听消息
            //channel.close();
            //connection.close();
        }
    }

## 三。程序运行效果 ##

先启动两个消费者，再启动生产者向消息队列中发送30条消息：

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDAwOTQ0Nw_size_16_color_FFFFFF_t_70 1]

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDAwOTQ0Nw_size_16_color_FFFFFF_t_70 2]

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDAwOTQ0Nw_size_16_color_FFFFFF_t_70 3]  
结论：

**在一个队列中如果有多个消费者，那么消费者之间对于同一个消息的关系是竞争的关系。**

## 四。模式总结 ##

**1、简单模式 HelloWorld**

一个生产者、一个消费者，不需要设置交换机（使用默认的交换机）

**2、工作队列模式 Work Queue**

一个生产者、多个消费者（竞争关系），不需要设置交换机（使用默认的交换机）

**3、发布订阅模式 Publish/subscribe**

需要设置类型为fanout的交换机，并且交换机和队列进行绑定，当发送消息到交换机后，交换机会将消息发送到绑定的队列

**4、路由模式 Routing**

需要设置类型为direct的交换机，交换机和队列进行绑定，并且指定routing key，当发送消息到交换机后，交换机会根据routing key将消息发送到对应的队列

**5、通配符模式 Topic**

需要设置类型为topic的交换机，交换机和队列进行绑定，并且指定通配符方式的routing key，当发送消息到交换机后，交换机会根据routing key将消息发送到对应的队列

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDAwOTQ0Nw_size_16_color_FFFFFF_t_70]: /images/20221120/091b20752f814ccab472e693f75cfa08.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDAwOTQ0Nw_size_16_color_FFFFFF_t_70 1]: /images/20221120/3340816dd880450da797d338c10ac5b1.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDAwOTQ0Nw_size_16_color_FFFFFF_t_70 2]: /images/20221120/75adc311e2a549658d44e6c263f7bc56.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDAwOTQ0Nw_size_16_color_FFFFFF_t_70 3]: /images/20221120/d3cfae2b5293410aa3597e402951f599.png