RabbitMQ 公平分发解决消费者宕机消息丢失问题

我会带着你远行 2023-08-17 16:56 126阅读 0赞

工作队列-公平分发  
3.1模型  
生产者将消息放入消息队列，有两个以上的消费者进行消费。  
和轮训分发一样的是依旧拥有两个或两个以上的消费者，但是会根据消费者的处理消息能力分发任务  
假设C1的处理能力强，那么C1将会处理更多的任务，而C2会处理更少的任务。  
![在这里插入图片描述][20190929215534787.png]  
第一步：消费者关闭自动应答

channel.basicConsume(QUEUE_NAME,false ,consumer);

第二部：改成手动反馈处理完成的消息。即处理一次消息反馈一个回应

channel.basicAck(envelope.getDeliveryTag(),false);

第三部：限制每次发生数量，这样rabbitmq就不会一股脑的将数据全部分配好再发送了

channel.basicQos(1);

在消费者C1下添加以上代码

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);
            channel.basicQos(1);
            //定义一个消费者
            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);
                    channel.basicAck(envelope.getDeliveryTag(),false);
                    try { 
                        Thread.sleep(1000);
                    } catch (InterruptedException e) { 
                        e.printStackTrace();
                    }
                }
            };
           boolean autoack = false;
            //监听队列
            channel.basicConsume(QUEUE_NAME,autoack,consumer);
        }
    }

消费者C2与C1一样，为了体现C2处理能力较弱，唯一区别就是C2的睡眠时间的为2s。以下为运行结果  
![在这里插入图片描述][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]  
**从图上可以看出，由于C2的处理能力较弱,C1的处理能力较强，所以C1处理的消息比C2多。两者结束的时间也是差不多一致的。**  
下图即展示，关闭自动应答，限制每次发生数量，添加手动应答后消息不会因为消费者宕机而丢失的情况（为了让数据更有说明性发送数据改为了1000条）  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dzZGZ5bQ_size_16_color_FFFFFF_t_70 2]  
当消费者处理一段时间后，直接把消费者进程都关闭，可以明显的看到横左边时间从22:15：30到22:16：40消息数量依旧并未减少依旧保存在消息队列中。足以说明不会因为消费者宕机或者其他意外情况导致消息丢失  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dzZGZ5bQ_size_16_color_FFFFFF_t_70 3]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dzZGZ5bQ_size_16_color_FFFFFF_t_70 4]  
总结一下：  
channel.basicQos(n)；rabbitmq第一次分发会发N条消息给消费者,当消费者处理完一条消息后并且确认了某条消息之后 RabbitMQ 将相应的计数减1，之后消费者可以继续接收消息，直到再次到达计数上限  
1.关闭自动应答;  
2.改成手动反馈处理完成的消息;  
2.限制每次发生数量channel.basicQos(n);  
现象：能者多劳，处理快的可以得到更多的消息进行处理

这里说明一下channel.basicQos(n);  
举例说明，消费端程序调用了 channel.basicQos(5) ，之后订阅了某个队列进行消费。 RabbitM 会保存一个消费者的列表，每发送一条消息都会为对应的消费者计数，计数达到5后，那么RabbitMQ就不会向这个消费者再发消息。消费者确认了某条消息处理完后，RabbitMQ 将相应的计数减1之后消费者可以继续接收消息，直到再次到达计数上限。这种机制可以类比于 TCP IP中的"滑动窗口"

[20190929215534787.png]: /images/20230809/add85bf3397342fdb661a1453387b764.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dzZGZ5bQ_size_16_color_FFFFFF_t_70]: /images/20230809/0c5a1a3d3dbe4c5d929aae8212b3c6fa.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dzZGZ5bQ_size_16_color_FFFFFF_t_70 1]: /images/20230809/9d332a9822154727bd38308787723d2f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dzZGZ5bQ_size_16_color_FFFFFF_t_70 2]: /images/20230809/ea8f7355decc4a059d1538bf660386bd.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dzZGZ5bQ_size_16_color_FFFFFF_t_70 3]: /images/20230809/c10c2db0751e4819ae92fd56a4fac128.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dzZGZ5bQ_size_16_color_FFFFFF_t_70 4]: /images/20230809/8deabb580e544c4a805f9e1c95a9255d.png