Unfair distribution

In the beginning, we learned RabbitMQ The message is distributed in rotation , But in some scenarios, this strategy is not very good , For example, there are two consumers dealing with tasks , One of the consumers 1 Processing tasks is very fast , And another consumer 2

Processing speed is slow , At this time, we still use the rotation training distribution, so that the fast processing consumer is idle for a large part of the time , The slow consumer has been working , This distribution method is actually not very good in this case , however

RabbitMQ Not knowing this, it is still fairly distributed .

To avoid that , We can set parameters channel.basicQos(1);

public class UnfairWork {
    public   final static String QUEUE_NAME="hello3";
    public static void main(String[] args) throws Exception {

        System.out.println("c1 Short response ....");
        Channel channel = untils.getChannel();
        /**
         *  Unfair distribution 
         *  Set parameters 
         */
        int prefetchCount=1;
        channel.basicQos(1);
        /**
         *  Consumer information 
         * 1. Which queue to consume 
         * 2. Whether to answer automatically after successful consumption ,true Auto answer ,false manual 
         * 3. The callback content of unsuccessful consumption by consumers 1
         * 4. Consumer canceled callback 
         *
         */
        // Statement   receive messages 
        DeliverCallback deliverCallback=(consumerTag, delivery)->{

            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            //1. Message tags 
            // 2.false  It means that only the received transmitted information is answered ,true To answer all messages, including the delivered messages 
            channel.basicAck(delivery.getEnvelope().getDeliveryTag(),false);
            System.out.println(" Messages received "+new String(delivery.getBody()));
        };
        // Cancel     Callback of message 
        CancelCallback cancelCallback= consumerTag -> {
            System.out.println(consumerTag+" Message consumer interrupt ");
        };
        channel.basicConsume(QUEUE_NAME,false,deliverCallback,cancelCallback);


    }
}

 

/**
 *  This is the same code as the round robin 
 */
public class UnfairProduce {
    public static  final  String QUEUE_NAME="hello3";
    public static void main(String[] args) throws IOException, TimeoutException {

        Channel channel = untils.getChannel();
        channel.queueDeclare(QUEUE_NAME,false,false,false,null);
        // Receive messages from control too 
        Scanner scanner = new Scanner(System.in);
        while(scanner.hasNext())
        {
            String message=scanner.next();
            channel.basicPublish("",QUEUE_NAME,null,message.getBytes(StandardCharsets.UTF_8));;
            System.out.println(" Send message complete "+message);
        }

    }
}