消费端限流

当用户的请求突然增多，MQ可以配置消费端限流，让消息按照限制的数量进行消费，达到限流的效果。

下面我们来代码案例演示。

案例

我们接着在前面篇章的工程来进行代码演示即可。

1.配置 消费端限流

• 首先需要确保ack机制为手动确认 acknowledge="manual"
• 配置消费端每次消费的数量 prefetch="1"

<!--  定义监听器与队列的绑定  -->
<rabbit:listener-container connection-factory="connectionFactory" acknowledge="manual" prefetch="1">
    <rabbit:listener ref="ackListener" queue-names="test_queue_confirm"/>
</rabbit:listener-container>

2.编写监听器类

package com.lijw.listener;

import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
import org.springframework.stereotype.Component;


/**
 * Consumer 限流机制
 *  1. 确保ack机制为手动确认。
 *  2. listener-container配置属性
 *      perfetch = 1,表示消费端每次从mq拉去一条消息来消费，直到手动确认消费完毕后，才会继续拉去下一条消息。
 *
 * @author Aron.li
 * @date 2022/3/4 23:36
 */
@Component
public class QosListener implements ChannelAwareMessageListener {

    @Override
    public void onMessage(Message message, Channel channel) throws Exception {

        //1. 接收消息
        System.out.println(new String(message.getBody()));

        //2. 处理业务逻辑
        System.out.println("处理业务逻辑...");
        Thread.sleep(1000);

        //3. 手动签收
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), true);

    }

}

3.绑定 监听器类 与 队列

<!--  定义监听器与队列的绑定  -->
<rabbit:listener-container connection-factory="connectionFactory" acknowledge="manual" prefetch="1">
    <rabbit:listener ref="qosListener" queue-names="test_queue_confirm"/>
</rabbit:listener-container>

4.测试 能否正常接收到消息

首先我们在生产者端发送10条消息：

我们在消费端查看接收的消息情况：

5.测试拒绝 Ack 确认消息

上面的测试中，我们在消费端签收了消息，所以消息在逐条处理，如果我们不签收的话，那么将会卡在当前消息上，不会继续往下取消息：

生产端再发送10条消息，再次重启消费端：

从上面的效果来看，已经达到了限流效果了，只要前面的消息没有处理完毕，则不会取更多的消息来处理。

消费端限流小结

- 在<rabbit:listener-container> 中配置 prefetch属性设置消费端一次拉取多少消息
- 消费端的确认模式一定为手动确认。acknowledge="manual"