栏目分类:
子分类:
返回
名师互学网用户登录
快速导航关闭
当前搜索
当前分类
子分类
实用工具
热门搜索
名师互学网 > IT > 软件开发 > 后端开发 > Java

rabbitmq qos(rabbitmq connection)

Java 更新时间: 发布时间: IT归档 最新发布 模块sitemap 名妆网 法律咨询 聚返吧 英语巴士网 伯小乐 网商动力

rabbitmq qos(rabbitmq connection)

出现大量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();
      //这里要对异常情况进行进行处理
    }
}

}

控制台队列情况查看


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

计算公式

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}}
]
}
]

转载请注明:文章转载自 www.mshxw.com
本文地址:https://www.mshxw.com/it/773298.html
我们一直用心在做
关于我们 文章归档 网站地图 联系我们

版权所有 (c)2021-2022 MSHXW.COM

ICP备案号:晋ICP备2021003244-6号