消费端限流
什么是消费端限流
- 假设一个场景, 首先, 我们RabbitMQ服务器有上万条消息未处理的消息, 我们随机打开一个消费者客户端, 会出现下面情况
- 巨量的消息瞬间全部推送过来, 但是我们单个客户端无法同时处理这么多数据
- RabbitMQ提供了一种Qos(服务质量保证)功能, 即在非自动确认消息的前提下, 如果一定数目的消息(通过基于consumer或者channel设置Qos的值)未被确认前, 不进行消费新的消息
- void BasicQos(uint prefetchSize, ushort prefetchCount, bool global);
- 参数解释
- prefetchSize: 0
- prefetchCount: 会告诉RabbitMQ不要同时给一个消费者推送多余N个消息, 即一旦有N个消息还没有ACK, 则该consumer将block掉, 直到有消息ACK
- global: true\false 是否将上面设置应用于channel, 简单点说, 就是上面限制是channel级别还是consumer级别
- 注意
- prefetchSize和global这两项, rabbitmq没有实现, 暂且不研究
- prefetch_count和no_ack=false的情况下生效, 即在自动应答的情况下这两个值是不生效的
消费端限流代码实现
帮助类新增函数
public static AMQP.Queue.DeclareOk queueDeclare(Channel channel, String queueName, boolean durable) throws IOException {
return channel.queueDeclare(queueName, durable, false, false, null);
}
消费者
package com.dance.redis.mq.rabbit.qos;
import com.dance.redis.mq.rabbit.RabbitMQHelper;
import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.concurrent.TimeUnit;
public class Receiver {
public static void main(String[] args) throws Exception {
Channel channel = RabbitMQHelper.getChannel();
String queueName = "test001";
// durable 是否持久化消息
RabbitMQHelper.queueDeclare(channel,queueName,true);
channel.basicQos(0, 1, false);
Consumer consumer = new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag,
Envelope envelope,
AMQP.BasicProperties properties,
byte[] body)
throws IOException {
System.out.println("receive message:" + new String(body) + ", RoutingKey: " + envelope.getRoutingKey());
try {
TimeUnit.SECONDS.sleep(1);
} catch (InterruptedException e) {
e.printStackTrace();
}
channel.basicAck(envelope.getDeliveryTag(), false);
}
};
// 参数：队列名称、是否自动ACK、Consumer
channel.basicConsume(queueName, false, consumer);
// 等待回调函数执行完毕之后，关闭资源。
TimeUnit.SECONDS.sleep(50);
channel.close();
RabbitMQHelper.closeConnection();
}
}
生产者
package com.dance.redis.mq.rabbit.qos;
import com.dance.redis.mq.rabbit.RabbitMQHelper;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import java.util.HashMap;
import java.util.Map;
public class Sender {
public static void main(String[] args) throws Exception {
Channel channel = RabbitMQHelper.getChannel();
String queueName = "test001";
RabbitMQHelper.queueDeclare(channel, queueName, true);
Map<String, Object> headers = new HashMap<>();
AMQP.BasicProperties props = new AMQP.BasicProperties.Builder()
.deliveryMode(2)
.contentEncoding("UTF-8")
.headers(headers).build();
for (int i = 0; i < 5; i++) {
String msg = "Hello World RabbitMQ " + i;
channel.basicPublish("", queueName, props, msg.getBytes());
}
}
}
测试
启动消费者
启动生产者
查看消费者
