【RabbitMQ】RabbitMQ 学习教程（四）消费端限流策略

柔光的暖阳◎ 2023-10-03 15:44 35阅读 0赞

在进入今天的主题之前，咱们先来了解下一个方法 `channel.basicQos()` 方法

### 1. `channel.basicQos()` 方法 ###

channel.basicQos()：设置客户端最多接收未被 ACK 的消息的个数

【注意】：消费者端要把自动确认 autoAck 设置为 false，channel.basicQos() 方法才会有效果

它有三个重载方法：

basicQos(int prefetchCount);
    basicQos(int prefetchCount, boolean global);
    basicQos(int prefetchSize, int prefetchCount, boolean global);

各个参数的含义如下：

*  prefetchSize：可接收消息的大小。如果设置为 0 ，那么表示对消息本身的大小不限制
 *  prefetchCount：处理消息最大的数量。例如：如果输入3，那么可以最多有3个消息不应答，如果到达了3个，则发送端发给这个接收方得消息只会在队列中，而接收方不会有接收到消息的事件产生
 *  global：是不是针对整个 Connection 的，因为一个 Connection 可以有多个 Channel，如果是 false，则说明只是针对于这个 Channel 的

> 场景：一个生产者生产了 20 条消息，有两个消费者：Consumer、Consumer2。并将这两个消费者的prefetchCount 值设置成 5，Consumer 进行了 ACK，而 Consumer2 没有进行 ACK

生产者 Producer 代码：

public class Producer {
        
    
        public static void main(String[] args) {
        
            // 1. 获取连接
            Connection connection = null;
            try {
        
                connection = RabbitMqUtil.getConnection("生产者");
            } catch (Exception e) {
        
                System.out.println("获取连接时，出现异常");
            }
    
            Channel channel = null;
            try {
        
                // 2. 通过连接获取通道 Channel
                channel = connection.createChannel();
                String queueName = "code_simple_queue1";
                // 3. 通过通道创建声明队列
                channel.queueDeclare(queueName, false, false, false, null);
                // 4. 发送消息给队列 Queue
                for (int i = 0; i < 20; i++) {
        
                    // 5. 准备消息内容
                    String message = "Hello RabbitMQ" + i;
                    channel.basicPublish("", queueName, null, message.getBytes());
                    System.out.println("消息发送完成~~~发送的消息为：" + message);
                }
            } catch (Exception e) {
        
                e.printStackTrace();
            } finally {
        
                try {
        
                    RabbitMqUtil.close(connection, channel);
                } catch (Exception e) {
        
                    System.out.println("关闭时，出现异常");
                }
            }
        }
    }

消费者 Consumer 代码如下：

public class Consumer {
        
    
        public static void main(String[] args) throws Exception{
        
            // 获取连接
            Connection connection = RabbitMqUtil.getConnection("消费者");
    
            final Channel channel = connection.createChannel();
            String queueName = "code_simple_queue1";
    
            // 定义消费者
            com.rabbitmq.client.Consumer consumer = new DefaultConsumer(channel) {
        
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
        
    
                    // 消息id，mq 在 channel 中用来标识消息的 id，可用于确认消息已接收
                    long deliveryTag = envelope.getDeliveryTag();
                    // body 消息体
                    String msg = new String(body,"utf-8");
                    System.out.println("收到消息：" + msg);
                    /**
                     * @param1：deliveryTag:用来标识消息的id
                     * @param2：multiple：是否批量。true:将一次性 ACK 所有小于 deliveryTag 的消息
                     */
                    // 手动确认
                    channel.basicAck(deliveryTag, false);
                }
            };
    
            // 限制未 ACK 的消息最大数
            channel.basicQos(5, false);
            // 监听队列  手动 ACK
            channel.basicConsume(queueName, false, consumer);
    
            System.out.println("开始接收消息~~~");
            System.in.read();
    
            // 关闭信道、连接
            RabbitMqUtil.close(connection, channel);
        }
    }

消费者 Consumer2 代码如下：  
Consumer2 代码与 Consumer 代码及其相似，唯一不同点就是没有进行手动 ACK

// 定义消费者
    com.rabbitmq.client.Consumer consumer = new DefaultConsumer(channel) {
        
        @Override
        public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
        
    
            // 消息id，mq 在 channel 中用来标识消息的 id，可用于确认消息已接收
            long deliveryTag = envelope.getDeliveryTag();
            // body 消息体
            String msg = new String(body,"utf-8");
            System.out.println("收到消息：" + msg);
            /**
             * @param1：deliveryTag:用来标识消息的id
             * @param2：multiple：是否批量。true:将一次性 ACK 所有小于 deliveryTag 的消息
             */
            // 手动确认
            //channel.basicAck(deliveryTag, false);
        }
    };

表示，它只消费消息，并未确认消息。（这种情况会导致消息被重复消费哈！！）

【注意】：这两个消费者设置了最大未确认数是 5 （`channel.basicQos(5, false);`），当未确认消息数超过 5 就不会再接收消息了。

程序运行结果如下：

消费者 Consumer 消费消息情况：  
![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAc2NvNTI4Mg_size_20_color_FFFFFF_t_70_g_se_x_16]  
消费者 Consumer 消费了 15 条消息，因为它每消费一条就 ACK 一条，另外 5 条消息被另一个消费者 Consumer2 给持有

消费者 Consumer2 消费消息情况：  
![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAc2NvNTI4Mg_size_20_color_FFFFFF_t_70_g_se_x_16 1]  
Consumer2 消费了 5 条消息，但它都没有确认，由于 `channel.basicQos(5, false);` 的缘故，它就没再消费消息了。（Consumer2 ）

到 RabbitMQ 页面查看：  
![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAc2NvNTI4Mg_size_20_color_FFFFFF_t_70_g_se_x_16 2]  
发现该队列有 5 条消息未被确认。

### 2. 限流策略 ###

> 场景：高并发情况下，队列里面一瞬间就就积累了上万条数据，但是消费者无法同时处理这么多请求，这个时候当我们打开客户端，瞬间就有巨量的信息给推送过来、但是客户端是没有办法同时处理这么多数据的，结果就是消费者（客户端）挂掉了…

这种场景下我们就需要对消费端进行限流：`channel.basicQos();`

生产者 Product：

消费者 Consumer：

public class Consumer {
        
    
        public static void main(String[] args) throws Exception{
        
            // 获取连接
            Connection connection = RabbitMqUtil.getConnection("消费者");
    
            final Channel channel = connection.createChannel();
            String queueName = "code_simple_queue1";
    
            // 定义消费者
            com.rabbitmq.client.Consumer consumer = new DefaultConsumer(channel) {
        
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
        
    
                    // 消息id，mq 在 channel 中用来标识消息的 id，可用于确认消息已接收
                    long deliveryTag = envelope.getDeliveryTag();
                    // body 消息体
                    String msg = new String(body,"utf-8");
                    System.out.println("收到消息：" + msg);
                    /**
                     * @param1：deliveryTag:用来标识消息的id
                     * @param2：multiple：是否批量。true:将一次性 ACK 所有小于 deliveryTag 的消息
                     */
                    // 手动确认
                    channel.basicAck(deliveryTag, false);
                }
            };
    
            // 监听队列  手动 ACK
            channel.basicConsume(queueName, false, consumer);
    
            System.out.println("开始接收消息~~~");
            System.in.read();
    
            // 关闭信道、连接
            RabbitMqUtil.close(connection, channel);
        }
    }

消费者 Consumer2：

public class Consumer2 {
        
    
        public static void main(String[] args) throws Exception{
        
            // 获取连接
            Connection connection = RabbitMqUtil.getConnection("消费者");
    
            final Channel channel = connection.createChannel();
            String queueName = "code_simple_queue1";
    
            // 定义消费者
            com.rabbitmq.client.Consumer consumer = new DefaultConsumer(channel) {
        
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
        
    
                    // 消息id，mq 在 channel 中用来标识消息的 id，可用于确认消息已接收
                    long deliveryTag = envelope.getDeliveryTag();
                    // body 消息体
                    String msg = new String(body,"utf-8");
                    System.out.println("收到消息：" + msg);
                    /**
                     * @param1：deliveryTag:用来标识消息的id
                     * @param2：multiple：是否批量。true:将一次性 ACK 所有小于 deliveryTag 的消息
                     */
                    // 手动确认
                    channel.basicAck(deliveryTag, false);
                }
            };
    		
    		/**  设置限流机制
             *  param1: prefetchSize，消息本身的大小 如果设置为0  那么表示对消息本身的大小不限制
             *  param2: prefetchCount，告诉rabbitmq不要一次性给消费者推送大于N个消息
             *  param3：global，是否将上面的设置应用于整个通道，false表示只应用于当前消费者
             */
            // 限制未 ACK 的消息最大数
            channel.basicQos(0, 5, false);
            // 监听队列  手动 ACK
            channel.basicConsume(queueName, false, consumer);
    
            System.out.println("开始接收消息~~~");
            System.in.read();
    
            // 关闭信道、连接
            RabbitMqUtil.close(connection, channel);
        }
    }

[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAc2NvNTI4Mg_size_20_color_FFFFFF_t_70_g_se_x_16]: https://img-blog.csdnimg.cn/8f52fc8d6a254a43acf5bc95f8a8bed3.png?x-oss-process=image/watermark,type_ZHJvaWRzYW5zZmFsbGJhY2s,shadow_50,text_Q1NETiBAc2NvNTI4Mg==,size_20,color_FFFFFF,t_70,g_se,x_16
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAc2NvNTI4Mg_size_20_color_FFFFFF_t_70_g_se_x_16 1]: https://img-blog.csdnimg.cn/776098d3148f4dc5b2e9eca85c52434b.png?x-oss-process=image/watermark,type_ZHJvaWRzYW5zZmFsbGJhY2s,shadow_50,text_Q1NETiBAc2NvNTI4Mg==,size_20,color_FFFFFF,t_70,g_se,x_16
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAc2NvNTI4Mg_size_20_color_FFFFFF_t_70_g_se_x_16 2]: https://img-blog.csdnimg.cn/57b54b43806f4aafb262975d68709dea.png?x-oss-process=image/watermark,type_ZHJvaWRzYW5zZmFsbGJhY2s,shadow_50,text_Q1NETiBAc2NvNTI4Mg==,size_20,color_FFFFFF,t_70,g_se,x_16