RabbitMQ 工作队列

系统管理员 2023-08-17 16:31 129阅读 0赞

## 工作队列 ##

**2.1模型**  
生产者将消息放入消息队列，有两个以上的消费者进行消费。  
![在这里插入图片描述][20190928231837231.png]  
为什么会有工作队列  
上文介绍了simple队列，一一对应的；而且生产者发送消息很快，不费时间；消费者需要处理业务，花费时间较多；消息队列容易挤压大量消息；  
这个时候便可以增加消费者并且与相应队列进行绑定再进行消息的消费，减少消息的挤压。  
**2.2增加消费者C1,C2**

public class RecvC1 { 
        private final static String QUEUE_NAME = "work_queue";
    
        public static void main(String[] argv) throws Exception { 
    
            // 获取到连接以及mq通道
            Connection connection = ConnectionUtil.getConnection();
            // 从连接中创建通道
            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 { 
                    super.handleDelivery(consumerTag, envelope, properties, body);
                    String msg = new String(body,"utf-8");
                    System.out.println("rec[1]:"+msg);
                    try { 
                        Thread.sleep(1000);
                    } catch (InterruptedException e) { 
                        e.printStackTrace();
                    }
                }
            };
            //监听队列
            channel.basicConsume(QUEUE_NAME,true,consumer);
        }
    }

public class RecvC2 { 
        private final static String QUEUE_NAME = "work_queue";
    
        public static void main(String[] argv) throws Exception { 
    
            // 获取到连接以及mq通道
            Connection connection = ConnectionUtil.getConnection();
            // 从连接中创建通道
            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 { 
                    super.handleDelivery(consumerTag, envelope, properties, body);
                    String msg = new String(body,"utf-8");
                    System.out.println("rec[2]:"+msg);
                    try { 
                        Thread.sleep(2000);
                    } catch (InterruptedException e) { 
                        e.printStackTrace();
                    }
                }
            };
            //监听队列
            channel.basicConsume(QUEUE_NAME,true,consumer);
        }
    }

**2.3生产者**

public class Send { 
        private final static String QUEUE_NAME = "work_queue";
    
        public static void main(String[] argv) throws Exception { 
            // 获取到连接以及mq通道
            Connection connection = ConnectionUtil.getConnection();
            // 从连接中创建通道
            Channel channel = connection.createChannel();
            // 声明（创建）队列
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
            // 消息内容
            for (int i=0;i<1000;i++){ 
                String message = "Hello World!"+i;
                channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
                Thread.sleep(20);
                System.out.println(" [x] Sent '" + message + "'");
            }
            //关闭通道和连接
            channel.close();
            connection.close();
        }
    }

这个时候打开C1,C2消费者，再打开生产者（一定要先打开消费者再打开生产者）。可以看到C1C2的情况如下。**可以看到C1消费的都是偶数，C2消费的是奇数。并且在消息还没有消费完，然而打开rabbitmq管理界面却看到消息队列中已经没有消息了**下面开始解释为什么会产生这种情况。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dzZGZ5bQ_size_16_color_FFFFFF_t_70]![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dzZGZ5bQ_size_16_color_FFFFFF_t_70 1]  
![在这里插入图片描述][2019092823534820.png]  
由下图当C1工作完后，C2还在工作。并且C1C2处理的工作量是一样的。C2的处理消息能力较弱。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dzZGZ5bQ_size_16_color_FFFFFF_t_70 2]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dzZGZ5bQ_size_16_color_FFFFFF_t_70 3]  
为什么会产生这种情况：当一个队列中存在多个消费者，那么队列在将消息发送给消费者时采用的轮训分发：消息队列在分配消息时不用等消息处理完，**就会将消息C1一个C2一个再C1一个**（所以C1得到的消息数量和C2一样多）。**并且不管消费者什么时候处理完消息就一股脑的将消息全部分发出去**，所以就会出现rabbitmq中消息已经分发完成，但是消费者还没有处理完消息。

## 轮询分发的缺点： ##

*  消息堆积：每个消费者得到的消息数量一样，但是处理任务的能力不一样，就会导致有的消费者处理完消息，有的消费者还没处理完消息，造成消息堆积。
 *  消息丢失，将一个消费者一个消息均等分配好，再全部发送消息。发送完这个时候消息队列的消息已经消失了，如果这个时候消费者宕机或者其他意外情况，那些还没来得及处理的消息便会丢失。

## 解决方案 ##

公平分发：[RabbitMQ 公平分发 解决消费者宕机消息丢失问题 https://blog.csdn.net/wsdfym/article/details/101722108][RabbitMQ _ _ https_blog.csdn.net_wsdfym_article_details_101722108]

[20190928231837231.png]: /images/20230810/e3703f35f8c5430cafbd81bc14acd9df.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dzZGZ5bQ_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/2019092823514877.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dzZGZ5bQ==,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dzZGZ5bQ_size_16_color_FFFFFF_t_70 1]: /images/20230810/b06b343e9e704580a85e3f40ee22f3a5.png
[2019092823534820.png]: /images/20230810/269f3c5c3dee47ee911440f88392ee41.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dzZGZ5bQ_size_16_color_FFFFFF_t_70 2]: /images/20230810/642d56fc82034d7186a204354b5052c6.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dzZGZ5bQ_size_16_color_FFFFFF_t_70 3]: /images/20230810/31d5481e22174867a983f58994690935.png
[RabbitMQ _ _ https_blog.csdn.net_wsdfym_article_details_101722108]: https://blog.csdn.net/wsdfym/article/details/101722108