RabbitMQ —— Round-robin 轮询分发

怼烎@ 2023-02-15 01:28 1阅读 0赞

# 前文 #

[消息中间件 —— 简介][Link 1]

[RabbitMQ —— 介绍][RabbitMQ _]

[RabbitMQ —— 下载、安装][RabbitMQ _ 1]

[RabbitMQ —— 工作模式][RabbitMQ _ 2]

[RabbitMQ —— 简单队列][RabbitMQ _ 3]

### 文章目录 ###

*  前文
 *  简单队列的不足
 *  WorkerQueue（工作队列）
 *  连接配置
 *  生产者
 *  消费者
 *  测试
 *  总结

# 简单队列的不足 #

耦合性高，生产者一一对应消费者（如果想有多个消费者消费队列中的消息就不 OK 了）

队列名变更的话，这时候得同时变更（耦合性高）

# WorkerQueue（工作队列） #

什么是工作队列的呢？就是一个生产者对应多个消费者  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dvb19ob21l_size_16_color_FFFFFF_t_70]

*  P ： 表示生产者
 *  中间的：表示队列
 *  C ： 表示消费者

为什么会出现工作队列？

Simple 队列是一一对应的，而且我们实际开发中，生产者发送消息是毫不费力的，而消费者一般是要跟业务相结合的，消费者接收到消息之后就需要处理，可能需要花费时间，这时候队列就会积压了很多消息

# 连接配置 #

package com.java.rabbitmq.util;
    
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.ConnectionFactory;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    /** * @author Woo_home * @create 2020/5/26 11:18 */
    
    public class ConnectionUtils { 
    
        /** * 获取 MQ 的连接 * @return */
        public static Connection getConnection() throws IOException, TimeoutException { 
            // 定义一个连接工厂
            ConnectionFactory factory = new ConnectionFactory();
    
            // 设置服务地址
            factory.setHost("localhost");
    
            // 设置端口号(AMQP)
            factory.setPort(5672);
    
            // 设置 virtual hosts
            factory.setVirtualHost("/vhost");
    
            // 设置用户名
            factory.setUsername("admin");
    
            // 设置密码
            factory.setPassword("admin");
    
            return factory.newConnection();
        }
    }

# 生产者 #

package com.java.rabbitmq.worker;
    
    import com.java.rabbitmq.util.ConnectionUtils;
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    
    /** * @author Woo_home * @create 2020/5/26 13:06 */
    
    public class ProducerDemo { 
    
        /** * F ---> Queue ---> C1,C2 * @param args * @throws Exception */
    
        // 队列名称
        private static final String QUEUE_NAME = "test_worker_queue";
    
        public static void main(String[] args) throws Exception { 
    
            // 获取连接
            Connection connection = ConnectionUtils.getConnection();
    
            // 获取 Channel
            Channel channel = connection.createChannel();
    
            // 声明队列
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
    
            // 往队列里面发送
            for (int i = 0; i < 50; i++) { 
                String msg = "Hello RabbitMQ " + i;
                channel.basicPublish("", QUEUE_NAME, null, msg.getBytes());
                System.out.println("send msg : " + msg);
                Thread.sleep(i * 20);
            }
    
            // 关闭资源
            channel.close();
            connection.close();
        }
    }

# 消费者 #

package com.java.rabbitmq.worker;
    
    import com.java.rabbitmq.util.ConnectionUtils;
    import com.rabbitmq.client.*;
    import java.io.IOException;
    
    /** * @author Woo_home * @create 2020/5/26 13:12 */
    
    public class ConsumerDemo { 
    
        // 队列名称
        private static final String QUEUE_NAME = "test_worker_queue";
    
        public static void main(String[] args) throws Exception { 
            // 获取连接
            Connection connection = ConnectionUtils.getConnection();
    
            // 获取 channel
            Channel channel = connection.createChannel();
    
            // 声明队列
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
    
            // 定义一个消费者
            DefaultConsumer consumer = new DefaultConsumer(channel) { 
                // 消息到达触发这个方法
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { 
                    String s = new String(body, "utf-8");
                    System.out.println("msg : " + s);
    
                    try { 
                        Thread.sleep(1000);
                    } catch (InterruptedException e) { 
                        e.printStackTrace();
                    } finally { 
                        System.out.println("done");
                    }
                }
            };
    
            // 监听队列
            channel.basicConsume(QUEUE_NAME, true, consumer);
        }
    }

# 测试 #

先启动消息生产者，运行如下：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dvb19ob21l_size_16_color_FFFFFF_t_70 1]  
刷新 Queue 界面 [http://localhost:15672/\#/queues][http_localhost_15672_queues]  
![在这里插入图片描述][20200526134125315.png]  
然后再启动两个消费者，如果不能同时启动两个一样的程序的话，进行如下配置即可：

点击 Edit Configurations  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dvb19ob21l_size_16_color_FFFFFF_t_70 2]  
选择你要运行两次的程序，都选 share 然后点击 apply 即可  
![在这里插入图片描述][20200526134559460.png]  
程序输出：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dvb19ob21l_size_16_color_FFFFFF_t_70 3]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dvb19ob21l_size_16_color_FFFFFF_t_70 4]

# 总结 #

消费者 1 消费的是 偶数，消费者 2 消费的是奇数，或者反过来，这种方式叫做轮询分发（Round-Robin），结果就是不管谁忙或者谁清闲们都不会多给一个消息

任务消息总是你一个我一个

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

相关 MQ 文章阅读

[ActiveMQ 下载、安装][ActiveMQ]

[ActiveMQ —— Java 连接 ActiveMQ（点对点）][ActiveMQ _ Java _ ActiveMQ]

[ActiveMQ —— Java 连接 ActiveMQ（发布订阅 Topic）][ActiveMQ _ Java _ ActiveMQ_ Topic]

[ActiveMQ —— Broker][ActiveMQ _ Broker]

[ActiveMQ —— Spring 整合 ActiveMQ][ActiveMQ _ Spring _ ActiveMQ]

[SpringBoot 整合 ActiveMQ][SpringBoot _ ActiveMQ]

[Link 1]: https://blog.csdn.net/Woo_home/article/details/105894485
[RabbitMQ _]: https://blog.csdn.net/Woo_home/article/details/106354749
[RabbitMQ _ 1]: https://blog.csdn.net/Woo_home/article/details/106349501
[RabbitMQ _ 2]: https://blog.csdn.net/Woo_home/article/details/106355966
[RabbitMQ _ 3]: https://blog.csdn.net/Woo_home/article/details/106540970
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dvb19ob21l_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20200526130146391.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dvb19ob21l,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dvb19ob21l_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/2020052613225718.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dvb19ob21l,size_16,color_FFFFFF,t_70
[http_localhost_15672_queues]: http://localhost:15672/#/queues
[20200526134125315.png]: https://img-blog.csdnimg.cn/20200526134125315.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dvb19ob21l_size_16_color_FFFFFF_t_70 2]: https://img-blog.csdnimg.cn/20200526134502253.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dvb19ob21l,size_16,color_FFFFFF,t_70
[20200526134559460.png]: https://img-blog.csdnimg.cn/20200526134559460.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dvb19ob21l_size_16_color_FFFFFF_t_70 3]: https://img-blog.csdnimg.cn/20200526134201781.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dvb19ob21l,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dvb19ob21l_size_16_color_FFFFFF_t_70 4]: https://img-blog.csdnimg.cn/20200526134213718.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dvb19ob21l,size_16,color_FFFFFF,t_70
[ActiveMQ]: https://blog.csdn.net/Woo_home/article/details/105895137
[ActiveMQ _ Java _ ActiveMQ]: https://blog.csdn.net/Woo_home/article/details/105895628
[ActiveMQ _ Java _ ActiveMQ_ Topic]: https://blog.csdn.net/Woo_home/article/details/105902770
[ActiveMQ _ Broker]: https://blog.csdn.net/Woo_home/article/details/106277802
[ActiveMQ _ Spring _ ActiveMQ]: https://blog.csdn.net/Woo_home/article/details/106278364
[SpringBoot _ ActiveMQ]: https://blog.csdn.net/Woo_home/article/details/106282637