RabbitMQ公平队列原理和使用

川长思鸟来 2022-01-28 13:43 261阅读 0赞

# [RabbitMQ公平队列原理和使用][RabbitMQ] #

目前消息转发机制是平均分配，这样就会出现俩个消费者，奇数的任务很耗时，偶数的任何工作量很小，造成的原因就是近当消息到达队列进行转发消息。并不在乎有多少任务消费者并未传递一个应答给RabbitMQ。仅仅盲目转发所有的奇数给一个消费者，偶数给另一个消费者。

为了解决这样的问题，我们可以使用basicQos方法，传递参数为prefetchCount= 1。这样告诉RabbitMQ不要在同一时间给一个消费者超过一条消息。  
换句话说，只有在 消费者空闲的时候会发送下一条信息。调度分发消息的方式，也就是告诉RabbitMQ每次只给消费者处理一条消息，也就是等待消费者处理完毕并自己对刚刚处理的消息进行确认之后，才发送下一条消息，防止消费者太过于忙碌，也防止它太过去清闲。  
通过 设置channel.basicQos(1);

*![1179709-20181110210321197-1110304867.png][]*

服务器能力不同，能者多劳。 均摊模式的话，都处理相同数量的

消息队列 发出去的消息被消费完了 然后收到 ack包 才可以继续发给他

** 公平队列原理：队列服务器向消费者发送消息的时候，消费者采用手动应答模式，队列服务器必须要收到消费者发送ack结果通知，才会发送下一个消息。（快的处理的多，消费的多）**

producer：

package com.toov5.Producer;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.toov5.utils.MQConnectionUtils;
    
    public class Producer {
        // 队列名称
        private static final String UEUE_NAME = "test_queue";
    
        public static void main(String[] args) throws IOException, TimeoutException {
            // 创建新的连接
            Connection connection = MQConnectionUtils.newConnection();
            // 创建Channel
            Channel channel = connection.createChannel();
            // 创建队列
            channel.queueDeclare(UEUE_NAME, false, false, false, null);
            channel.basicQos(1); // 保证 取一个消费  队列给消费者发送消息时候 一个消息
            for (int i = 0; i < 10; i++) {
                // 创建message
                String msg = "toov5_message";
                System.out.println("生产者投递消息" + msg + i);
                // 生产者发送消息
                channel.basicPublish("", UEUE_NAME, null, msg.getBytes());
            }
            // 关闭通道和连接
            channel.close();
            connection.close();
    
        }
    }

Consumer1

package com.toov5.Consumer;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    import com.rabbitmq.client.AMQP.BasicProperties;
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.DefaultConsumer;
    import com.rabbitmq.client.Envelope;
    import com.toov5.utils.MQConnectionUtils;
    
    public class Consumer1 {
      
         //队列名称
            private static final String QUEUE_NAME = "test_queue";
            
            public static void main(String[] args) throws IOException, TimeoutException {
                System.out.println("消费者启动..........1");
                //创建新的连接
            Connection connection = MQConnectionUtils.newConnection();
               //创建Channel
                final Channel channel = connection.createChannel();
                // 消费者关联队列
                 channel.queueDeclare(QUEUE_NAME, false, false, false, null);
                 channel.basicQos(1);
                  DefaultConsumer defaultConsumerr = new DefaultConsumer(channel) {
                      //监听获取消息
                        @Override
                        public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties,
                                byte[] body) throws IOException {
                            String msg =new String(body,"UTF-8");
                            System.out.println("消费者获取生产者消息："+msg);
                            try {
                                //模拟应答等待时间
                                Thread.sleep(1000);
                            } catch (Exception e) {
                                
                            }finally {
                               channel.basicAck(envelope.getDeliveryTag(), false);  //手动应答 告诉消息队列服务器 消费成功
                            }
                        }
                  };
                //牵手模式设置  默认自动应答模式  true:自动应答模式  
                  channel.basicConsume(QUEUE_NAME, false, defaultConsumerr);//    fanse手动应答          
    
            }
    }

Consumer2

package com.toov5.Consumer;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    import com.rabbitmq.client.AMQP.BasicProperties;
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.DefaultConsumer;
    import com.rabbitmq.client.Envelope;
    import com.toov5.utils.MQConnectionUtils;
    
    public class Consumer2 {
      
         //队列名称
            private static final String QUEUE_NAME = "test_queue";
            
            public static void main(String[] args) throws IOException, TimeoutException {
                System.out.println("消费者启动..........2");
                //创建新的连接
            Connection connection = MQConnectionUtils.newConnection();
               //创建Channel
                final Channel channel = connection.createChannel();
                // 消费者关联队列
                 channel.queueDeclare(QUEUE_NAME, false, false, false, null);
                  channel.basicQos(1);
                  DefaultConsumer defaultConsumerr = new DefaultConsumer(channel) {
                      //监听获取消息
                        @Override
                        public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties,
                                byte[] body) throws IOException {
                            String msg =new String(body,"UTF-8");
                            System.out.println("消费者获取生产者消息："+msg);
                            try {
                                //模拟应答等待时间
                                Thread.sleep(300);
                            } catch (Exception e) {
                                
                            }finally {
                            channel.basicAck(envelope.getDeliveryTag(), false);  //手动应答 告诉消息队列服务器 消费成功
                            }
                        }
                  };
                //牵手模式设置  默认自动应答模式  true:自动应答模式  
                  channel.basicConsume(QUEUE_NAME, false, defaultConsumerr);//    fanse手动应答          
            }
    }

运行结果：

睡眠少的（执行快的） 指定的多

![1179709-20181110222156636-1213648152.png][]

![1179709-20181110222207598-573938502.png][]

![1179709-20181110222222071-533565428.png][]

** 注意 每个消费者 必须要应答 一下！ 队列服务器没有收到应答 就不会发送下一个给消费者~**

[RabbitMQ]: https://www.cnblogs.com/toov5/p/9940873.html
[1179709-20181110210321197-1110304867.png]: /images/20220128/f3481f04f5d647b5843e470e51cde889.png
[1179709-20181110222156636-1213648152.png]: /images/20220128/7acdbba3cc124c658545ae9e982916d1.png
[1179709-20181110222207598-573938502.png]: /images/20220128/f9d0e83a464b4d3d8daf30e97cdc4f39.png
[1179709-20181110222222071-533565428.png]: /images/20220128/cfc0ff4c438e4068b153369cf006d8f3.png