消息队列 RabbitMQ[八] SpringBoot Consumer 限流机制

消息队列 RabbitMQ[一] RabbitMQ的下载与安装
消息队列 RabbitMQ[二] RabbitMQ可视化管控台创建用户并为用户分配虚拟机
消息队列 RabbitMQ[三] RabbitMQ的HelloWorld工作模式(SpringBoot方式与amqp-client方式)
消息队列 RabbitMQ[四] RabbitMQ的Publish/Subscribe(发布/订阅)工作模式(SpringBoot方式与amqp-client方式)
消息队列 RabbitMQ[五] RabbitMQ的Routing工作模式(SpringBoot方式与amqp-client方式)
消息队列 RabbitMQ[六] RabbitMQ的Topics工作模式(SpringBoot方式与amqp-client方式)
消息队列 RabbitMQ[七] RabbitMQ保证消息的可靠性传递（Confirm Return Ack）
消息队列 RabbitMQ[八] SpringBoot Consumer 限流机制
消息队列 RabbitMQ[九] SpringBoot 设置消息过期时间TTL
消息队列 RabbitMQ[十] SpringBoot 死信队列与延迟队列实现思路

比如在一个整点秒杀活动中，在某一时刻可能会产生大量的请求到服务器，这时服务器可能会因为到来的大量请求而崩溃，流量销峰就是将这些请求都放到rabbitmq中，然后服务器一次取承受范围之内个请求，这就会大大减少服务器的压力。

rabbitmq提供了限流机制来控制每次消费者取多少消息

先在topic_queue2中存储5条消息

import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class RabbitMQConfig {

    @Autowired
    private CachingConnectionFactory connectionFactory;

    @Bean(name = "consumerContainerFactory ")
    public SimpleRabbitListenerContainerFactory consumerContainerFactory(){
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        factory.setPrefetchCount(1); // 这里就是控制每次取一条消息類
        return factory;
    }
}

	@RabbitListener(queues = "direct_queue2", containerFactory = "consumerContainerFactory")
    public void listenQueueFanoutQ2(Message message) throws InterruptedException {
        System.out.println(new String(message.getBody()));
        Thread.sleep(3000);
    }

每隔3秒就会取出一条消息