SpringBoot 系列之集成 RabbitMQ 实现高效流量控制

系列博客专栏：

JVM系列博客专栏
SpringBoot系列博客

Spring Boot 2.2.1 集成 RabbitMQ 实现高效流量控制

在分布式系统中，消息队列是实现异步通信、解耦服务的重要组件。RabbitMQ 作为一款成熟的开源消息队列，广泛应用于各类项目中。本文将结合 Spring Boot 2.2.1，详细介绍如何集成 RabbitMQ 并实现基于队列长度、内存和磁盘的流量控制，同时引入服务端限流配置，进一步提升系统的稳定性与可靠性。

一、RabbitMQ 流量控制的重要性

当消息产生速度过快，超过消息队列的处理能力时，可能会导致队列积压、系统性能下降甚至崩溃。通过流量控制，可以有效限制消息的流入速度，使系统能够在合理的负载下运行，保障服务的稳定性和可靠性。

二、Spring Boot 2.2.1 集成 RabbitMQ 基础配置

1. 引入依赖

在 pom.xml 文件中添加 Spring Boot AMQP 和 Web 依赖：

<dependencies>
    <!-- Spring Boot Starter AMQP -->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-amqp</artifactId>
    </dependency>
    <!-- Spring Boot Starter Web -->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
    <!-- JSON处理依赖 -->
    <dependency>
        <groupId>com.fasterxml.jackson.core</groupId>
        <artifactId>jackson-databind</artifactId>
    </dependency>
    <!-- Spring Boot Starter Test -->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-test</artifactId>
        <scope>test</scope>
    </dependency>
    <!-- RabbitMQ测试依赖 -->
    <dependency>
        <groupId>org.springframework.amqp</groupId>
        <artifactId>spring-rabbit-test</artifactId>
        <scope>test</scope>
    </dependency>
</dependencies>

2. 配置文件

在 application.yml 中配置 RabbitMQ 连接信息和相关参数：

spring:
  rabbitmq:
    host: localhost
    port: 5672
    username: guest
    password: guest
    virtual-host: /
    requested-heartbeat: 30
    connection-timeout: 10000
    publisher-confirms: true
    publisher-returns: true
    listener:
      simple:
        acknowledge-mode: auto
        prefetch: 50
        concurrency: 3
        max-concurrency: 10
    cache:
      channel:
        size: 50
        checkout-timeout: 30000
      connection:
        mode: CHANNEL
        size: 5

# 自定义流量控制配置
app:
  flow-control:
    max-messages: 1000
    duration: 5000

3. RabbitMQ 配置类

创建 RabbitMQConfig 类，配置队列、交换机、绑定关系、消息转换器以及 RabbitTemplate：

package com.example.springboot.rabbitmq.configuration;

import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
@Slf4j
public class RabbitMQConfig {

    public static final String QUEUE_NAME = "flow.control.queue";
    public static final String EXCHANGE_NAME = "flow.control.exchange";
    public static final String ROUTING_KEY = "flow.control.key";

    // 配置队列
    @Bean
    public Queue queue() {
        return QueueBuilder.durable(QUEUE_NAME)
                .maxLength(1000)
                .build();
    }

    // 配置交换机
    @Bean
    public DirectExchange exchange() {
        return new DirectExchange(EXCHANGE_NAME);
    }

    // 绑定队列和交换机
    @Bean
    public Binding binding(Queue queue, DirectExchange exchange) {
        return BindingBuilder.bind(queue).to(exchange).with(ROUTING_KEY);
    }

    // 配置消息转换器
    @Bean
    public Jackson2JsonMessageConverter messageConverter() {
        return new Jackson2JsonMessageConverter();
    }

    // 配置RabbitTemplate
    @Bean
    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory,
                                         Jackson2JsonMessageConverter messageConverter) {
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        rabbitTemplate.setMessageConverter(messageConverter);
        
        // 设置mandatory标志，确保消息在无法路由时返回
        rabbitTemplate.setMandatory(true);
        
        // 设置发布确认回调
        rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
            if (ack) {
                log.info("消息发送成功: {}",  correlationData);
            } else {
                log.warn("消息发送失败: {}",  cause);
            }
        });
        
        // 设置返回回调
        rabbitTemplate.setReturnCallback((message, replyCode, replyText, exchange, routingKey) -> {
            log.info("消息被退回: {}", new String(message.getBody()));
            log.info("回复码: ", replyCode);
            log.info("回复文本: ", replyText);
            log.info("交换机: ", exchange);
            log.info("路由键: ", routingKey);
        });
        
        return rabbitTemplate;
    }

    // 配置监听器容器工厂
    @Bean
    public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory(
            ConnectionFactory connectionFactory,
            Jackson2JsonMessageConverter messageConverter) {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        factory.setMessageConverter(messageConverter);
        factory.setConcurrentConsumers(3); // 设置并发消费者数量
        factory.setMaxConcurrentConsumers(10);
        factory.setPrefetchCount(50); // 设置 QoS
        factory.setAcknowledgeMode(AcknowledgeMode.MANUAL); // 手动确认模式
        return factory;
    }
}

三、基于队列长度的流量控制

在 MessageProducer 类中实现基于队列长度的流量控制逻辑：

package com.example.demo.service;

import com.example.demo.config.RabbitMQConfig;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.support.CorrelationData;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.UUID;
import java.util.concurrent.atomic.AtomicInteger;

@Service
public class MessageProducer {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    private final AtomicInteger messageCount = new AtomicInteger(0);
    private static final int MAX_MESSAGES = 1000;
    private volatile boolean flowControlEnabled = false;

    public void sendMessage(String message) {
        if (flowControlEnabled) {
            System.out.println("流量控制已启用，暂停发送消息");
            return;
        }

        if (messageCount.get() >= MAX_MESSAGES) {
            System.out.println("达到最大消息数量，触发流量控制");
            enableFlowControl(5000);
            return;
        }

        String correlationId = UUID.randomUUID().toString();
        CorrelationData correlationData = new CorrelationData(correlationId);

        rabbitTemplate.convertAndSend(
                RabbitMQConfig.EXCHANGE_NAME,
                RabbitMQConfig.ROUTING_KEY,
                message,
                correlationData
        );

        messageCount.incrementAndGet();
        System.out.println("发送消息: " + message + ", 消息ID: " + correlationId);
    }

    public void enableFlowControl(long durationMillis) {
        flowControlEnabled = true;
        System.out.println("流量控制已启用，持续时间: " + durationMillis + "ms");

        new Thread(() -> {
            try {
                Thread.sleep(durationMillis);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
            flowControlEnabled = false;
            messageCount.set(0);
            System.out.println("流量控制已禁用");
        }).start();
    }
}

除了用代码限制外，可以用maxLength设置，示例代码：

 // 配置队列
 @Bean
 public Queue queue() {
     return QueueBuilder.durable(QUEUE_NAME)
             .maxLength(1000)
             .build();
 }

四、x-max-length-bytes 参数详解

x-max-length-bytes 用于限制队列中消息的总字节数。在创建队列时，可以通过代码配置：

@Bean
public Queue queue() {
    return QueueBuilder.durable(QUEUE_NAME)
           .maxLength(1000)
           .maxLengthBytes(1024 * 1024 * 10) // 设置队列消息总字节数上限为10MB
           .build();
}

当队列中消息的总字节数达到设定的阈值时，后续新消息的处理策略由 x-overflow 参数决定：

drop-head：丢弃队列头部的消息，为新消息腾出空间。
reject-publish：拒绝接收新消息，并向生产者返回 Basic.Reject 响应。

五、基于内存和磁盘的流量控制

通过配置 RabbitMQ 服务器的内存和磁盘告警阈值，当服务器内存使用或磁盘空间达到阈值时，会自动触发流量控制。例如：

rabbitmqctl set_vm_memory_high_watermark 0.6

此命令将内存高水位线设置为系统内存的 60%。

六、服务端限流配置

1. 基于 Guava 的限流实现

添加 Guava 依赖：

<dependency>
    <groupId>com.google.guava</groupId>
    <artifactId>guava</artifactId>
    <version>28.2-jre</version>
</dependency>

使用 RateLimiter 进行限流：

package com.example.demo.service;

import com.google.common.util.concurrent.RateLimiter;
import org.springframework.stereotype.Service;

@Service
public class LimitedService {

    private final RateLimiter rateLimiter = RateLimiter.create(5);

    public void limitedMethod() {
        if (rateLimiter.tryAcquire()) {
            System.out.println("请求被处理");
        } else {
            System.out.println("请求被限流");
        }
    }
}

七、消费端限流

默认情况下，如果不进行配置，RabbitMQ会尽可能快速地把队列中的消息发送到消费者。如果消息数量过多，可能会导致OOM或者影响其他进程的正常运行

1. 消费端限流示例

package com.example.springboot.rabbitmq.service;


import com.example.springboot.rabbitmq.configuration.RabbitMQConfig;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.retry.annotation.Backoff;
import org.springframework.retry.annotation.Retryable;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.lang.management.ManagementFactory;
import java.lang.management.OperatingSystemMXBean;

@Service
@Slf4j
public class MessageConsumer {

    @RabbitListener(queues = RabbitMQConfig.QUEUE_NAME)
    @Retryable(value = {IOException.class}, maxAttempts = 3,
            backoff = @Backoff(delay = 2000, multiplier = 2))
    public void receiveMessage(Message message, Channel channel) throws IOException {
        try {
            if (channel == null || !channel.isOpen()) {
                log.warn("Channel is closed or null, unable to process message");
                return;
            }
            // 动态设置预取计数
            channel.basicQos(calculatePrefetchCount());

            String content = new String(message.getBody());
            log.info("接收到消息:{} ", content);

            // 模拟消息处理时间
            Thread.sleep(100);

            // 发送消息确认
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);

            log.info("消息处理完成");
        } catch (Exception e) {
            log.error("处理消息时发生错误: {}", e.getMessage(), e);
            if (channel != null && channel.isOpen()) {
                channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true); // 失败后重新入队
            }
        }
    }


    // 根据系统负载动态计算预取计数
    private int calculatePrefetchCount() {
        double cpuLoad = getSystemCpuLoad();
        int basePrefetch = 10;
        return (int) Math.max(1, basePrefetch * (1 - cpuLoad));
    }

    // 获取当前系统 CPU 负载
    private double getSystemCpuLoad() {
        OperatingSystemMXBean osBean = (OperatingSystemMXBean) ManagementFactory.getOperatingSystemMXBean();
        return osBean.getSystemLoadAverage() / osBean.getAvailableProcessors();
    }

}