写一个spring boot的rabitmq限流demo

以下是一个使用Spring Boot和RabbitMQ的限流示例：

1. 首先，需要在pom.xml文件中添加以下依赖项：

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

2. 创建一个RabbitMQ配置类，用于定义连接工厂和RabbitMQ模板：

@Configuration
public class RabbitMQConfig {

    @Value("${spring.rabbitmq.host}")
    private String host;

    @Value("${spring.rabbitmq.port}")
    private int port;

    @Value("${spring.rabbitmq.username}")
    private String username;

    @Value("${spring.rabbitmq.password}")
    private String password;

    @Value("${spring.rabbitmq.virtual-host}")
    private String virtualHost;

    @Bean
    public ConnectionFactory connectionFactory() {
        CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
        connectionFactory.setHost(host);
        connectionFactory.setPort(port);
        connectionFactory.setUsername(username);
        connectionFactory.setPassword(password);
        connectionFactory.setVirtualHost(virtualHost);
        return connectionFactory;
    }

    @Bean
    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        return rabbitTemplate;
    }
}

3. 创建一个消息接收者，并使用@RabbitListener注释指定要监听的队列：

@Component
public class MessageReceiver {

    @RabbitListener(queues = "test.queue")
    public void receiveMessage(String message) {
        System.out.println("Received message: " + message);
    }
}

4. 创建一个消息发布者，并使用RabbitTemplate发送消息：

@Component
public class MessageSender {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    public void sendMessage(String message) {
        rabbitTemplate.convertAndSend("test.exchange", "test.routingkey", message);
    }
}

5. 创建一个限流过滤器，用于限制消息的发送速率：

@Component
public class RateLimitFilter implements Filter {

    private static final int MAX_MESSAGES_PER_SECOND = 10;
    private static final int INTERVAL = 1000;

    private final AtomicLong lastTime = new AtomicLong(System.currentTimeMillis());
    private final AtomicInteger count = new AtomicInteger(0);

    @Override
    public boolean preHandle(HttpServletRequest request, HttpServletResponse response, Object handler) throws Exception {
        long currentTime = System.currentTimeMillis();
        long timeSinceLast = currentTime - lastTime.get();
        if (timeSinceLast > INTERVAL) {
            int currentCount = count.getAndSet(0);
            lastTime.set(currentTime);
            return true;
        }
        else {
            int currentCount = count.incrementAndGet();
            if (currentCount > MAX_MESSAGES_PER_SECOND) {
                response.setStatus(HttpStatus.TOO_MANY_REQUESTS.value());
                return false;
            }
            else {
                return true;
            }
        }
    }
}

6. 将限流过滤器添加到应用程序中：

@Configuration
public class WebConfig implements WebMvcConfigurer {

    @Autowired
    private RateLimitFilter rateLimitFilter;

    @Override
    public void addInterceptors(InterceptorRegistry registry) {
        registry.addInterceptor(rateLimitFilter);
    }
}

7. 在应用程序中使用消息发布者发送消息：

@Autowired
private MessageSender messageSender;

@GetMapping("/send")
public void sendMessage() {
    messageSender.sendMessage("Hello, RabbitMQ!");
}

现在，应用程序将限制每秒发送的消息数量，以避免过度负载RabbitMQ服务器。