RabbitMQ不公平分发问题分析及问题解决

1.不公平分发

1.1 不公平分发策略是什么？

在 RabbitMQ 中，不公平分发（Unfair Dispatch）是指当多个消费者（Consumers）同时订阅同一个队列（Queue）时，消息的分发机制是不公平的，可能导致负载不均衡等问题。

1.2 不公平分发产生的原因？

默认情况下，RabbitMQ 采用的是轮询（Round Robin）的方式将消息（工作线程）平均分发给各个消费者，也就是理论上每一个消费者消费的消息数量都是一样的。但是实际上，可能由于消费者处理消息的速度不同(可能由于网络因素、配置因素等不同)，可能就会导致有些消费者长时间处于空闲状态，有些消费者消息处理不过来，导致消息积压，导致负载不均衡等情况，会严重影响到整个系统的性能。

1.3 怎么解决？

1.3.1 公平分发

公平分发：在公平预取模式下，每个消费者一次只能从队列中预先获取一条消息。当消费者处理完当前的消息并确认后，RabbitMQ 才会将下一条消息发送给该消费者，没有处理完消息的消费者将继续处理消息。这样可以确保每个消费者都能够公平地接收和处理消息，避免某个消费者长时间占用资源导致负载不均衡。

代码演示公平分发策略，在消费者中消费消息之前，设置参数 channel.basicQos(1);

public class Work05 {//队列名称public static final String TASK_QUEUE_NAME = "ACK_QUEUE";//接受消息public static void main(String[] args) throws IOException, TimeoutException {Channel channel = RabbitMQUtils.getChannel();System.out.println("C1等待接受消息处理时间较短");DeliverCallback deliverCallback = (consumerTag, message) -> {//沉睡1SSleepUtils.sleep(1);System.out.println("接受到的消息:" + new String(message.getBody(), "UTF-8"));//手动应答/*** 1.消息的标记Tag* 2.是否批量应答 false表示不批量应答信道中的消息*/channel.basicAck(message.getEnvelope().getDeliveryTag(), false);};CancelCallback cancelCallback = (consumerTag -> {System.out.println(consumerTag + "消费者取消消费接口回调逻辑");});//设置不公平分发int prefetchCount = 1;channel.basicQos(prefetchCount);//采用手动应答boolean autoAck = false;channel.basicConsume(TASK_QUEUE_NAME, autoAck, deliverCallback, cancelCallback);}
}

开启成功，会看到如下结果：

演示结果如下

C2消费者处理消息时间较长，在处理消息完成之前，不会收到新的消息。C1消费者处理消息效率较高，因为处理完一条消息后将继续处理收到的新消息。

1.3.2 预取值分发

预取值分发策略：带权的消息分发，公平分发策略虽然解决了负载不均衡的问题，但是在高并发场景下会存在性能问题。

试想一下，每个消费者消息消息时都会从一个缓存区里面读取消息，这个缓存区有大小限制，既解决了消息积压的问题，又优化了消费者处理完消息之后才向队列中读取一条数据这种损耗性能的操作。

因此可以通过一个未确认的消息缓冲区，开发人员能限制此缓冲区的大小，以避免缓冲区里面无限制的未确认消息问题。这个时候也是可以通过使用 basic.qos 方法设置「预取计数」值来完成的。

该值定义通道上允许的未确认消息的最大数量。一旦数量达到配置的数量， RabbitMQ 将停止在通道上传递更多消息，除非至少有一个未处理的消息被确认，例如，假设在通道上有未确认的消息 5、6、7，8，并且通道的预取计数设置为 4，此时 RabbitMQ 将不会在该通道上再传递任何消息，除非至少有一个未应答的消息被 ack。比方说 tag=6 这个消息刚刚被确认 ACK，RabbitMQ 将会感知这个情况到并再发送一条消息。消息应答和 QoS 预取值对用户吞吐量有重大影响。

通常，增加预取将提高向消费者传递消息的速度。虽然自动应答传输消息速率是最佳的，但是，在这种情况下已传递但尚未处理的消息的数量也会增加，从而增加了消费者的 RAM 消耗(随机存取存储器)应该小心使用具有无限预处理的自动确认模式或手动确认模式，消费者消费了大量的消息如果没有确认的话，会导致消费者连接节点的内存消耗变大，所以找到合适的预取值是一个反复试验的过程，不同的负载该值取值也不同 100 到 300 范围内的值通常可提供最佳的吞吐量，并且不会给消费者带来太大的风险。

预取值为 1 是最保守的。当然这将使吞吐量变得很低，特别是消费者连接延迟很严重的情况下，特别是在消费者连接等待时间较长的环境中。对于大多数应用来说，稍微高一点的值将是最佳的。

public class Work03 {//队列名称public static final String TASK_QUEUE_NAME = "ACK_QUEUE";//接受消息public static void main(String[] args) throws IOException, TimeoutException {Channel channel = RabbitMQUtils.getChannel();System.out.println("C1等待接受消息处理时间较短");DeliverCallback deliverCallback =(consumerTag,message) ->{//沉睡1SSleepUtils.sleep(1);System.out.println("接受到的消息:"+new String(message.getBody(),"UTF-8"));//手动应答/*** 1.消息的标记Tag* 2.是否批量应答 false表示不批量应答信道中的消息*/channel.basicAck(message.getEnvelope().getDeliveryTag(),false);};CancelCallback cancelCallback = (consumerTag -> {System.out.println(consumerTag + "消费者取消消费接口回调逻辑");});//设置不公平分发//int prefetchCount = 1;//值不等于 1，则代表预取值,预取值为4int prefetchCount = 4;channel.basicQos(prefetchCount);//采用手动应答boolean autoAck = false;channel.basicConsume(TASK_QUEUE_NAME,autoAck,deliverCallback,cancelCallback);}
}