当用户的请求突然增多,MQ可以配置消费端限流,让消息按照限制的数量进行消费,达到限流的效果。
下面我们来代码案例演示。
我们接着在前面篇章的工程来进行代码演示即可。
<!-- 定义监听器与队列的绑定 -->
<rabbit:listener-container connection-factory="connectionFactory" acknowledge="manual" prefetch="1">
<rabbit:listener ref="ackListener" queue-names="test_queue_confirm"/>
</rabbit:listener-container>
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);
}
}
<!-- 定义监听器与队列的绑定 -->
<rabbit:listener-container connection-factory="connectionFactory" acknowledge="manual" prefetch="1">
<rabbit:listener ref="qosListener" queue-names="test_queue_confirm"/>
</rabbit:listener-container>
首先我们在生产者端发送10条消息:
我们在消费端查看接收的消息情况:
上面的测试中,我们在消费端签收了消息,所以消息在逐条处理,如果我们不签收的话,那么将会卡在当前消息上,不会继续往下取消息:
生产端再发送10条消息,再次重启消费端:
从上面的效果来看,已经达到了限流效果了,只要前面的消息没有处理完毕,则不会取更多的消息来处理。
- 在<rabbit:listener-container> 中配置 prefetch属性设置消费端一次拉取多少消息
- 消费端的确认模式一定为手动确认。acknowledge="manual"