消费端限流:
什么是消费端限流?
场景:
我们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); }
原文地址:https://www.cnblogs.com/luhan777/p/11193148.html