3.rabbitmq轮询和不公平分发

rabbitmq轮询和不公平分发

rabbitmq轮询分发

rabbitmq默认是使用轮询来分发消息的。测试代码如下所示

生产者代码

/**
 * 生产者 task  rabbitmq 轮询演示
 */
public class Task {

    public static void main(String[] args) throws IOException {
        Channel channel = RabbitUtil.getChannel();
        channel.queueDeclare(QueueNames.HELLO,false,false,false,null);
        Scanner scanner = new Scanner(System.in);
        while (scanner.hasNext()){
            String message = scanner.next();
            channel.basicPublish("",QueueNames.HELLO,null,message.getBytes(StandardCharsets.UTF_8));
            System.out.println("task 发送消息完成，消息内容："+message);
        }
    }

}

消费者1代码

/**
 * 消费者01
 */
public class Worker01 {

    public static void main(String[] args) throws IOException {
        Channel channel = RabbitUtil.getChannel();
        DeliverCallback deliverCallback = (consumerTag, message) -> {
            String received = new String(message.getBody());
            System.out.println("接收到消息："+received);
        };
        CancelCallback cancelCallback = consumerTag -> {
            System.out.println(consumerTag+"消费者取消消费接口回调逻辑");
        };
        System.out.println("worker01 消费者正在等待消费");
        channel.basicConsume(QueueNames.HELLO,true,deliverCallback,cancelCallback);
    }
}

消费者2代码

/**
 * 消费者02
 */
public class Worker02 {

    public static void main(String[] args) throws IOException {
        Channel channel = RabbitUtil.getChannel();
        DeliverCallback deliverCallback = (consumerTag, message) -> {
            String received = new String(message.getBody());
            System.out.println("接收到消息："+received);
        };
        CancelCallback cancelCallback = consumerTag -> {
            System.out.println(consumerTag+"消费者取消消费消息的回调接口");
        };
        System.out.println("worker02 等待接收消息");
        channel.basicConsume(QueueNames.HELLO,true,deliverCallback,cancelCallback);
    }
}

• 启动生产者代码，多次输入消息。
• 启动多个消费者代码，发现多个消费者依次接收消息，且每条消息只执行一次

不公平分发

如果消息是轮询分发，如果每个消费者执行消费代码的时长不同，处理速度不同，那么很容易造成执行速度快的消费者会闲置，而执行比较慢的消费者会一直处在高消耗的状态。这样显然是不合理的，为了解决这个问题，我们可以采用不公平分发

在消费者中指定prefetch的值

Integer prefetch = 10
channel.basicQos(prefetch);

该值定义通道上允许的未确认消息的最大数量,即我这条通道允许未确认的消息数量是10个，一旦你分配给我的消息数量达到上限，那么rabbitmq就停止在这条通道上传递更多消息，转而去给别的通道分配消息。

perfetch的取值是根据业务来具体测试的，如果取值过高，那么内存中未确认的消息数量会很多，徒增内存消耗，如果取值太低，那么系统吞吐量下降，用户体验会很差。

生产者代码

/**
 * 不公平分发测试
 */
public class UnfairProducer {

    public static void main(String[] args) throws IOException {
        Channel channel = RabbitUtil.getChannel();
        channel.queueDeclare(QueueNames.UNFAIR,true,false,false,null);
        for (int i = 0; i < 100; i++) {
            String message = String.valueOf(i);
            channel.basicPublish("",QueueNames.UNFAIR, MessageProperties.PERSISTENT_TEXT_PLAIN,message.getBytes(StandardCharsets.UTF_8));
            System.out.println("生产者发出消息："+message);
        }
    }
}

消费者1代码

/**
 * 不公平消费者1
 */
public class UnfairConsumer1 {

    public static void main(String[] args) throws IOException {
        Channel channel = RabbitUtil.getChannel();
        System.out.println("c1 等待接收消息处理时间较短");
        //指定prefetch的值
        channel.basicQos(10);
        DeliverCallback deliverCallback = (consumerTag, message) -> {
            String received = new String(message.getBody());
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            System.out.println("c1 收到消息："+received);
            channel.basicAck(message.getEnvelope().getDeliveryTag(),false);
        };
        CancelCallback cancelCallback = consumerTag -> {
            System.out.println("消费者取消消费");
        };
        boolean autoAck = false;
        channel.basicConsume(QueueNames.UNFAIR,autoAck,deliverCallback,cancelCallback);
    }
}

消费者2代码

/**
 * 不公平消费者2
 */
public class UnfairConsumer2 {

    public static void main(String[] args) throws IOException {
        Channel channel = RabbitUtil.getChannel();
        channel.basicQos(2);
        System.out.println("c2 等待接收消息处理时间较短");
        DeliverCallback deliverCallback = (consumerTag, message) -> {
            String received = new String(message.getBody());
            try {
                Thread.sleep(10000);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            System.out.println("c2 收到消息："+received);
            channel.basicAck(message.getEnvelope().getDeliveryTag(),false);
        };
        CancelCallback cancelCallback = consumerTag -> {
            System.out.println("消费者取消消费");
        };
        Boolean autoAck = false;
        channel.basicConsume(QueueNames.UNFAIR,autoAck,deliverCallback,cancelCallback);
    }
}