RabbitMQ消费端限流策略(十)
2024-08-24 13:21:22
消费端限流:
什么是消费端限流?
场景:
我们RabbitMQ服务器有上万条未处理的消息,我们随便打开一个消费者客户端,会出现下面情况:
巨量的消息瞬间全部推送过来,但是我们单个客户端无法同时处理这么多数据。(导致服务器崩溃,线上故障)
生产端一次推送几百条数据库,客户端只接收一两条,在高并发的情况下,不能再生产端做限流,只能在消费端处理。
解决方法:
RabbitMQ提供了一种qos(服务质量保证)功能,在非自动确认消息的前提下,
如果一定数据的消息(通过基于consumer或者channel设置qos的值)未被确认前,不进行消费新的消息。减压减负
void BasicQos(uint prefetchSize,ushort prefetchCount,bool global);
消费端体现,一次最多能处理多少条消息(基本上为1),限流策略在什么上应用(channel--true,consumer---false)
prefetchSize:0
prefetchCount:会告诉RabbitMQ不要同时给一个消费者推送多余n个消息,
一旦有n个消息还没有ack,则该consumer将block调,知道有消息ack
global:true\false是否将上面设置应用于channel,简单的说就是上面限制是channel
级别的还是consumer级别,基本使用false。
注意:prefetchSize和global这两项,rabbitmq没有实现,暂不研究
prefetch_count在no_ack=false的情况下生效,在自动应答的情况下两个值不生效。
//生产端代码
ConnectionFactory connectionFactory = new ConnectionFactory();
connectionFactory.setHost("127.0.0.1");
connectionFactory.setPort(5672);
connectionFactory.setVirtualHost("/"); Connection connection = connectionFactory.newConnection();
Channel channel = connection.createChannel(); String exchange = "test_qos_exchange";
String routingKey = "qos.save"; String msg = "Hello RabbitMQ QOS Message"; for(int i =0; i<5; i ++){
channel.basicPublish(exchange, routingKey, true, null, msg.getBytes());
}
//消费端代码
ConnectionFactory connectionFactory = new ConnectionFactory();
connectionFactory.setHost("127.0.0.1");
connectionFactory.setPort(5672);
connectionFactory.setVirtualHost("/"); Connection connection = connectionFactory.newConnection();
Channel channel = connection.createChannel(); String exchangeName = "test_qos_exchange";
String queueName = "test_qos_queue";
String routingKey = "qos.#"; channel.exchangeDeclare(exchangeName, "topic", true, false, null);
channel.queueDeclare(queueName, true, false, false, null);
channel.queueBind(queueName, exchangeName, routingKey); //1 限流方式 第一件事就是 autoAck设置为 false
//接收1条消息,
channel.basicQos(0, 1, false); channel.basicConsume(queueName, false, new MyConsumer(channel));
//自定义消息端
private Channel channel ; public MyConsumer(Channel channel) {
super(channel);
//接收ack进行消息发送
this.channel = channel;
} @Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
System.err.println("-----------consume message----------");
System.err.println("consumerTag: " + consumerTag);
System.err.println("envelope: " + envelope);
System.err.println("properties: " + properties);
System.err.println("body: " + new String(body));
//false不支持批量签收
channel.basicAck(envelope.getDeliveryTag(), false); }
最新文章
- MySQL DML 整理
- javascript运算符与表达式
- git入门学习
- android 下载文件
- 从零开始学Bootstrap(1)
- UICollectionView使用
- 【python】python异常类型
- How to fix broken packages?(转)
- 20150303--从SQL中获取数据的三级联动
- Python Requests库:HTTP for Humans
- height为100%的问题
- JAVASCRIPT实现XML分页
- ZOJ 2048(Prim 或者 Kruskal)
- 关于WCF的引用,添加服务和添加web服务的区别
- Unity3d之剥离alpha通道
- FileSync文件同步更新工具
- 【译】第18节---数据注解-ForeignKey
- vue组件生命周期详解
- 读DataSnap源代码(四)
- The way to Go(6): Go程序的基本结构和要素