RabbitMq的阻塞与Qos关系_simplerabbitlistenercontainerfactory.setprefetchco

出现大量unacked

配置

@Bean
public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory() {
SimpleRabbitListenerContainerFactory simpleRabbitListenerContainerFactory = new SimpleRabbitListenerContainerFactory();
simpleRabbitListenerContainerFactory.setConnectionFactory(collectionFactory);
simpleRabbitListenerContainerFactory.setMessageConverter(new Jackson2JsonMessageConverter());
//消费者ACK应答模式,QOS只有再手动模式下才有效
simpleRabbitListenerContainerFactory.setAcknowledgeMode(AcknowledgeMode.MANUAL);
//消息拒绝接受，true标识消息重新入队，默认为true
simpleRabbitListenerContainerFactory.setDefaultRequeueRejected(true);
//默认是250
simpleRabbitListenerContainerFactory.setPrefetchCount(250);
return simpleRabbitListenerContainerFactory;
}

生产者A

@Test
public void pruductA() {
System.out.println(“pruductA开始发送消息”);
Queue queue = new Queue(“testA”, true);
//模拟100万条数据
for (int i = 0; i < 1000000; i++) {
rabbitTemplate.convertAndSend(“testA”, “testA” + String.valueOf(i + 1));
}
System.out.println(“pruductA消息发送结束”);
}

消费者A

@Component
@RabbitListener(queues = “testA”)
public class CustomerAMQListener {

@RabbitHandler
public void handler(String msg, Message message, Channel channel) {
    long msgTag = message.getMessageProperties().getDeliveryTag();
    String body = new String(message.getBody());
    System.out.println("cutomerA接收到的消息为：" + body);
    try {
        channel.basicAck(msgTag, false);
    } catch (IOException e) {
        e.printStackTrace();
      //这里要对异常情况进行进行处理
    }
}
1
2
3
4
5
6
7
8
9
10
11
12

}

控制台队列情况查看

从控制台可以看到发布消息的速度和客户端的处理消息的速度基本持平，这里不会阻塞

计算公式

min = concurrency* PrefetchCount* 消费节点数量
max = max-concurrency * PrefetchCount* 消费节点数量

什么情况下会阻塞？

unacked_msg_count < min 队列不会阻塞。但需要及时处理 unacked 的消息。
unacked_msg_count >= min 可能会出现堵塞。
unacked_msg_count >= max 队列一定阻塞。

为什么要使用Qos？

Qos是mq提供了 channel.basicQos该方法来限制一次消费的消息数量，注意Qos只有在手动确认ack下才生效，自动ack是不生效的，手动和自动，可以通过factory.setAcknowledgeMode(AcknowledgeMode.MANUAL)进行配置
如果完全不配置QoS，这样RabbitMQ会尽其所能快速地发送队列中的消息到消费端，而客户端的消费能力是有限的，极有可能导致客户端OOM而被打死。
Qos设置过小可能会导致mq队列阻塞
Qos设置过大则可能会导致客户端消费不过来而导致消息在内存中大量堆积，消耗大量内存
默认是250，需要进行实际测试，根据系统的能力和业务场景选择一个合适的值

Qos相关方法

channel.basicQos(0); // No limit for this consumer
channel.basicQos(10, false); // Per consumer limit
channel.basicQos(15, true); // Per channel limit
还可以通过配置文件指定默认值：
advanced.config file
[
{rabbit, [
{default_consumer_prefetch, {false,250}}
]
}
]