一、限流阈值设置的痛点

上个月，我在为一个电商系统做性能优化时，遇到了一个非常棘手的问题：

"我们的系统在高峰期经常出现限流误杀，而在低峰期又限流不足，"技术总监皱着眉头说，"固定的限流阈值根本无法适应业务的动态变化，我们需要一个智能的方案来自动调整限流阈值。"

我查看了他们的限流配置，发现问题确实很严重：

• 系统使用固定的限流阈值，无法适应流量的动态变化
• 高峰期阈值设置过低，导致正常请求被误杀
• 低峰期阈值设置过高，无法有效保护系统
• 无法根据历史流量数据进行智能调优
• 没有自动推荐合理的限流阈值的机制
• 限流策略缺乏灵活性和适应性

更关键的是，他们根本不知道如何设置一个合理的限流阈值，只能依靠经验和猜测。

二、传统方案的局限性

1. 固定阈值限流

使用固定的限流阈值，无论流量如何变化，都使用相同的限制。

// 固定阈值限流
@Bean
public RateLimiter rateLimiter() {
    return RateLimiter.create(100); // 固定100 QPS
}

这种方案的问题：

• 无法适应流量变化：无法根据流量的动态变化调整阈值
• 高峰期误杀：高峰期阈值设置过低，导致正常请求被误杀
• 低峰期保护不足：低峰期阈值设置过高，无法有效保护系统
• 缺乏灵活性：固定的阈值无法适应不同时间段的流量模式
• 依赖经验：阈值设置依赖于人工经验，缺乏数据支撑

2. 手动调整阈值

依靠运维人员根据经验手动调整限流阈值。

# 手动调整限流阈值
curl -X POST http://localhost:8080/api/rate-limit/set -d "qps=150"

这种方案的问题：

• 反应滞后：发现问题时通常已经造成了影响
• 效率低下：需要人工持续监控和调整
• 误判风险：人工判断容易出现误判
• 无法预测：无法提前预测流量变化并调整阈值
• 成本高昂：需要专门的运维人员进行监控和调整

3. 简单的时间分片

根据不同的时间段设置不同的限流阈值。

// 简单的时间分片限流
public int getCurrentQpsLimit() {
    LocalTime now = LocalTime.now();
    if (now.isAfter(LocalTime.of(8, 0)) && now.isBefore(LocalTime.of(10, 0))) {
        return 200; // 早高峰
    } else if (now.isAfter(LocalTime.of(12, 0)) && now.isBefore(LocalTime.of(14, 0))) {
        return 150; // 午高峰
    } else if (now.isAfter(LocalTime.of(18, 0)) && now.isBefore(LocalTime.of(22, 0))) {
        return 250; // 晚高峰
    } else {
        return 100; // 低峰期
    }
}

这种方案的问题：

• 缺乏灵活性：时间分片固定，无法适应突发流量
• 精度不足：时间分片粒度较粗，无法精确捕捉流量变化
• 无法自适应：无法根据实际流量情况自动调整
• 维护成本高：需要定期手动更新时间分片配置
• 无法应对异常：无法处理异常流量模式

三、终极方案：基于历史流量的限流阈值动态调优

今天，我要和大家分享一个在实战中验证过的解决方案：基于历史流量的限流阈值动态调优。

这套方案的核心思想是：

1. 实时监控：实时采集和分析系统的流量数据
2. 历史分析：基于历史流量数据进行统计和分析
3. 智能推荐：根据历史流量模式自动推荐合理的限流阈值
4. 动态调整：根据实时流量和历史数据动态调整限流阈值
5. 自适应学习：通过持续学习不断优化限流策略

四、方案详解

1. 核心原理

限流阈值动态调优的工作流程如下：

系统产生流量数据
    ↓
流量监控器实时采集数据
    ↓
数据处理层聚合分析
    ↓
历史数据分析（统计特征、模式识别）
    ↓
阈值推荐算法计算
    ↓
动态调整限流阈值
    ↓
限流执行器应用新阈值
    ↓
持续监控和优化

2. SpringBoot 实现

（1）添加 Maven 依赖

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-actuator</artifactId>
</dependency>
<dependency>
    <groupId>io.micrometer</groupId>
    <artifactId>micrometer-registry-prometheus</artifactId>
</dependency>
<dependency>
    <groupId>com.google.guava</groupId>
    <artifactId>guava</artifactId>
    <version>31.1-jre</version>
</dependency>
<dependency>
    <groupId>org.apache.commons</groupId>
    <artifactId>commons-math3</artifactId>
    <version>3.6.1</version>
</dependency>
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>

（2）流量监控服务

@Service
@Slf4j
public class TrafficMonitoringService {

    private static final String TRAFFIC_METRIC_NAME = "application.traffic.qps";
    private static final String ENDPOINT_TAG = "endpoint";

    private final MeterRegistry meterRegistry;
    private final RedisTemplate<String, Object> redisTemplate;

    private Map<String, Counter> endpointCounters = new ConcurrentHashMap<>();

    public TrafficMonitoringService(MeterRegistry meterRegistry, RedisTemplate<String, Object> redisTemplate) {
        this.meterRegistry = meterRegistry;
        this.redisTemplate = redisTemplate;
    }

    public void recordRequest(String endpoint) {
        // 记录请求计数
        Counter counter = endpointCounters.computeIfAbsent(endpoint, k ->
                Counter.builder(TRAFFIC_METRIC_NAME)
                        .tag(ENDPOINT_TAG, endpoint)
                        .register(meterRegistry)
        );
        counter.increment();

        // 记录到 Redis，用于历史数据分析
        String key = "traffic:endpoint:" + endpoint + ":" + LocalDate.now();
        String field = LocalTime.now().format(DateTimeFormatter.ofPattern("HH:mm"));
        redisTemplate.opsForHash().increment(key, field, 1);
    }

    public Map<String, Double> getCurrentQps() {
        Map<String, Double> qpsMap = new ConcurrentHashMap<>();
        // 从 Prometheus 或其他监控系统获取实时 QPS
        // 这里简化实现，实际项目中需要从监控系统查询
        return qpsMap;
    }

    public Map<String, List<Double>> getHistoricalTraffic(String endpoint, int days) {
        Map<String, List<Double>> trafficData = new ConcurrentHashMap<>();

        try {
            for (int i = 0; i < days; i++) {
                LocalDate date = LocalDate.now().minusDays(i);
                String key = "traffic:endpoint:" + endpoint + ":" + date;
                
                Map<Object, Object> hourlyData = redisTemplate.opsForHash().entries(key);
                List<Double> hourlyQps = new ArrayList<>();
                
                for (int hour = 0; hour < 24; hour++) {
                    double hourQps = 0;
                    for (int minute = 0; minute < 60; minute++) {
                        String field = String.format("%02d:%02d", hour, minute);
                        Object value = hourlyData.get(field);
                        if (value != null) {
                            hourQps += Double.parseDouble(value.toString());
                        }
                    }
                    hourlyQps.add(hourQps / 60); // 转换为 QPS
                }
                
                trafficData.put(date.toString(), hourlyQps);
            }
        } catch (Exception e) {
            log.error("Failed to get historical traffic data", e);
        }

        return trafficData;
    }

    public Map<String, TrafficStats> getTrafficStats(String endpoint, int days) {
        Map<String, TrafficStats> statsMap = new ConcurrentHashMap<>();
        Map<String, List<Double>> historicalTraffic = getHistoricalTraffic(endpoint, days);

        for (Map.Entry<String, List<Double>> entry : historicalTraffic.entrySet()) {
            String date = entry.getKey();
            List<Double> hourlyQps = entry.getValue();

            TrafficStats stats = new TrafficStats();
            stats.setDate(date);
            stats.setHourlyQps(hourlyQps);
            stats.setMaxQps(Collections.max(hourlyQps));
            stats.setMinQps(Collections.min(hourlyQps));
            stats.setAvgQps(hourlyQps.stream().mapToDouble(Double::doubleValue).average().orElse(0));
            stats.setP95Qps(calculatePercentile(hourlyQps, 95));
            stats.setP99Qps(calculatePercentile(hourlyQps, 99));

            statsMap.put(date, stats);
        }

        return statsMap;
    }

    private double calculatePercentile(List<Double> values, double percentile) {
        if (values.isEmpty()) {
            return 0;
        }
        List<Double> sortedValues = new ArrayList<>(values);
        Collections.sort(sortedValues);
        int index = (int) Math.ceil(percentile / 100.0 * sortedValues.size()) - 1;
        return sortedValues.get(Math.max(0, index));
    }

    @Data
    public static class TrafficStats {
        private String date;
        private List<Double> hourlyQps;
        private double maxQps;
        private double minQps;
        private double avgQps;
        private double p95Qps;
        private double p99Qps;
    }
}

（3）限流阈值推荐服务

@Service
@Slf4j
public class RateLimitRecommendationService {

    private static final int HISTORY_DAYS = 7; // 历史数据天数
    private static final double SAFETY_FACTOR = 1.2; // 安全系数
    private static final double SMOOTHING_FACTOR = 0.3; // 平滑因子

    @Autowired
    private TrafficMonitoringService trafficMonitoringService;

    public Map<String, Double> recommendRateLimits() {
        Map<String, Double> recommendations = new ConcurrentHashMap<>();

        // 获取所有端点的历史流量数据
        List<String> endpoints = getAllEndpoints();
        for (String endpoint : endpoints) {
            double recommendedLimit = recommendRateLimitForEndpoint(endpoint);
            recommendations.put(endpoint, recommendedLimit);
        }

        return recommendations;
    }

    public double recommendRateLimitForEndpoint(String endpoint) {
        try {
            Map<String, TrafficMonitoringService.TrafficStats> statsMap = 
                    trafficMonitoringService.getTrafficStats(endpoint, HISTORY_DAYS);

            if (statsMap.isEmpty()) {
                return 100; // 默认值
            }

            // 计算历史最大 QPS 和分位数
            List<Double> maxQpsList = new ArrayList<>();
            List<Double> p95QpsList = new ArrayList<>();
            List<Double> p99QpsList = new ArrayList<>();

            for (TrafficMonitoringService.TrafficStats stats : statsMap.values()) {
                maxQpsList.add(stats.getMaxQps());
                p95QpsList.add(stats.getP95Qps());
                p99QpsList.add(stats.getP99Qps());
            }

            // 计算基准值
            double maxQps = Collections.max(maxQpsList);
            double avgP95Qps = p95QpsList.stream().mapToDouble(Double::doubleValue).average().orElse(0);
            double avgP99Qps = p99QpsList.stream().mapToDouble(Double::doubleValue).average().orElse(0);

            // 综合计算推荐值
            double baseValue = Math.max(maxQps, Math.max(avgP95Qps, avgP99Qps));
            double recommendedLimit = baseValue * SAFETY_FACTOR;

            // 平滑处理
            double currentLimit = getCurrentRateLimit(endpoint);
            if (currentLimit > 0) {
                recommendedLimit = currentLimit * (1 - SMOOTHING_FACTOR) + recommendedLimit * SMOOTHING_FACTOR;
            }

            // 确保最小值
            recommendedLimit = Math.max(recommendedLimit, 10);

            log.info("Recommended rate limit for endpoint {}: {} QPS", endpoint, recommendedLimit);
            return recommendedLimit;

        } catch (Exception e) {
            log.error("Failed to recommend rate limit for endpoint {}", endpoint, e);
            return 100; // 默认值
        }
    }

    private List<String> getAllEndpoints() {
        // 获取所有端点
        // 实际项目中需要从配置或数据库中获取
        return Arrays.asList("/api/normal", "/api/high-traffic", "/api/critical");
    }

    private double getCurrentRateLimit(String endpoint) {
        // 获取当前限流阈值
        // 实际项目中需要从配置或缓存中获取
        return 100; // 默认值
    }

    public Map<String, Double> adjustRateLimitsBasedOnRealTime() {
        Map<String, Double> adjustments = new ConcurrentHashMap<>();
        Map<String, Double> currentQps = trafficMonitoringService.getCurrentQps();
        Map<String, Double> recommendations = recommendRateLimits();

        for (Map.Entry<String, Double> entry : currentQps.entrySet()) {
            String endpoint = entry.getKey();
            double currentQpsValue = entry.getValue();
            double recommendedLimit = recommendations.getOrDefault(endpoint, 100.0);

            // 根据实时流量动态调整
            double adjustmentFactor = 1.0;
            if (currentQpsValue > recommendedLimit * 0.8) {
                // 接近阈值，适当提高
                adjustmentFactor = 1.1;
            } else if (currentQpsValue < recommendedLimit * 0.3) {
                // 远低于阈值，适当降低
                adjustmentFactor = 0.9;
            }

            double adjustedLimit = recommendedLimit * adjustmentFactor;
            adjustments.put(endpoint, adjustedLimit);

            log.info("Adjusted rate limit for endpoint {}: {} QPS (current QPS: {})
",
                    endpoint, adjustedLimit, currentQpsValue);
        }

        return adjustments;
    }
}

（4）动态限流服务

@Service
@Slf4j
public class DynamicRateLimitService {

    private static final String RATE_LIMIT_KEY_PREFIX = "rate_limit:";

    private final RedisTemplate<String, Object> redisTemplate;
    private final RateLimitRecommendationService recommendationService;

    private Map<String, RateLimiter> rateLimiters = new ConcurrentHashMap<>();

    public DynamicRateLimitService(RedisTemplate<String, Object> redisTemplate,
                                 RateLimitRecommendationService recommendationService) {
        this.redisTemplate = redisTemplate;
        this.recommendationService = recommendationService;
    }

    public boolean tryAcquire(String endpoint) {
        RateLimiter rateLimiter = getRateLimiter(endpoint);
        return rateLimiter.tryAcquire();
    }

    public boolean tryAcquire(String endpoint, long timeout, TimeUnit unit) {
        RateLimiter rateLimiter = getRateLimiter(endpoint);
        return rateLimiter.tryAcquire(timeout, unit);
    }

    private RateLimiter getRateLimiter(String endpoint) {
        return rateLimiters.computeIfAbsent(endpoint, k -> {
            double rateLimit = getRateLimitFromRedis(endpoint);
            if (rateLimit <= 0) {
                rateLimit = recommendationService.recommendRateLimitForEndpoint(endpoint);
                saveRateLimitToRedis(endpoint, rateLimit);
            }
            return RateLimiter.create(rateLimit);
        });
    }

    public void updateRateLimit(String endpoint, double rateLimit) {
        RateLimiter rateLimiter = rateLimiters.get(endpoint);
        if (rateLimiter != null) {
            // 重新创建 RateLimiter
            rateLimiters.put(endpoint, RateLimiter.create(rateLimit));
        }
        saveRateLimitToRedis(endpoint, rateLimit);
        log.info("Updated rate limit for endpoint {}: {} QPS", endpoint, rateLimit);
    }

    public void updateAllRateLimits() {
        Map<String, Double> recommendations = recommendationService.recommendRateLimits();
        for (Map.Entry<String, Double> entry : recommendations.entrySet()) {
            String endpoint = entry.getKey();
            double rateLimit = entry.getValue();
            updateRateLimit(endpoint, rateLimit);
        }
    }

    public void adjustRateLimitsBasedOnRealTime() {
        Map<String, Double> adjustments = recommendationService.adjustRateLimitsBasedOnRealTime();
        for (Map.Entry<String, Double> entry : adjustments.entrySet()) {
            String endpoint = entry.getKey();
            double rateLimit = entry.getValue();
            updateRateLimit(endpoint, rateLimit);
        }
    }

    private double getRateLimitFromRedis(String endpoint) {
        try {
            String key = RATE_LIMIT_KEY_PREFIX + endpoint;
            Object value = redisTemplate.opsForValue().get(key);
            if (value != null) {
                return Double.parseDouble(value.toString());
            }
        } catch (Exception e) {
            log.error("Failed to get rate limit from Redis", e);
        }
        return 0;
    }

    private void saveRateLimitToRedis(String endpoint, double rateLimit) {
        try {
            String key = RATE_LIMIT_KEY_PREFIX + endpoint;
            redisTemplate.opsForValue().set(key, rateLimit);
        } catch (Exception e) {
            log.error("Failed to save rate limit to Redis", e);
        }
    }

    public Map<String, Double> getCurrentRateLimits() {
        Map<String, Double> rateLimits = new ConcurrentHashMap<>();
        for (Map.Entry<String, RateLimiter> entry : rateLimiters.entrySet()) {
            String endpoint = entry.getKey();
            double rateLimit = getRateLimitFromRedis(endpoint);
            if (rateLimit > 0) {
                rateLimits.put(endpoint, rateLimit);
            }
        }
        return rateLimits;
    }
}

（5）限流拦截器

@Component
@Slf4j
public class RateLimitInterceptor implements HandlerInterceptor {

    @Autowired
    private DynamicRateLimitService rateLimitService;

    @Autowired
    private TrafficMonitoringService trafficMonitoringService;

    @Override
    public boolean preHandle(HttpServletRequest request, HttpServletResponse response, Object handler) {
        String endpoint = request.getRequestURI();

        // 记录请求
        trafficMonitoringService.recordRequest(endpoint);

        // 尝试获取令牌
        if (!rateLimitService.tryAcquire(endpoint)) {
            response.setStatus(HttpStatus.TOO_MANY_REQUESTS.value());
            try {
                response.getWriter().write("Rate limit exceeded");
            } catch (IOException e) {
                log.error("Failed to write response", e);
            }
            return false;
        }

        return true;
    }
}

（6）限流监控定时任务

@Component
@Slf4j
public class RateLimitMonitoringTask {

    @Autowired
    private DynamicRateLimitService rateLimitService;

    @Scheduled(fixedRate = 3600000) // 每小时执行一次
    public void updateRateLimits() {
        try {
            rateLimitService.updateAllRateLimits();
            log.info("Updated rate limits based on historical data");
        } catch (Exception e) {
            log.error("Error updating rate limits", e);
        }
    }

    @Scheduled(fixedRate = 300000) // 每5分钟执行一次
    public void adjustRateLimitsBasedOnRealTime() {
        try {
            rateLimitService.adjustRateLimitsBasedOnRealTime();
            log.info("Adjusted rate limits based on real-time traffic");
        } catch (Exception e) {
            log.error("Error adjusting rate limits", e);
        }
    }
}

（7）配置类

@Configuration
public class RateLimitConfig implements WebMvcConfigurer {

    @Autowired
    private RateLimitInterceptor rateLimitInterceptor;

    @Override
    public void addInterceptors(InterceptorRegistry registry) {
        registry.addInterceptor(rateLimitInterceptor)
                .addPathPatterns("/api/**");
    }

    @Bean
    public RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory redisConnectionFactory) {
        RedisTemplate<String, Object> template = new RedisTemplate<>();
        template.setConnectionFactory(redisConnectionFactory);
        template.setKeySerializer(new StringRedisSerializer());
        template.setValueSerializer(new Jackson2JsonRedisSerializer<>(Object.class));
        return template;
    }

    @Bean
    public TrafficMonitoringService trafficMonitoringService(MeterRegistry meterRegistry, RedisTemplate<String, Object> redisTemplate) {
        return new TrafficMonitoringService(meterRegistry, redisTemplate);
    }

    @Bean
    public RateLimitRecommendationService rateLimitRecommendationService() {
        return new RateLimitRecommendationService();
    }

    @Bean
    public DynamicRateLimitService dynamicRateLimitService(RedisTemplate<String, Object> redisTemplate, RateLimitRecommendationService recommendationService) {
        return new DynamicRateLimitService(redisTemplate, recommendationService);
    }

    @Bean
    public RateLimitInterceptor rateLimitInterceptor() {
        return new RateLimitInterceptor();
    }

    @Bean
    public MeterRegistryCustomizer<MeterRegistry> metricsCommonTags() {
        return registry -> registry.config()
                .commonTags("application", "rate-limit-demo");
    }

    // 内部类，用于 MeterRegistry 自定义
    private interface MeterRegistryCustomizer<T> {
        void customize(T t);
    }
}

（8）配置文件

spring:
  application:
    name: rate-limit-demo
  redis:
    host: localhost
    port: 6379
    password:
    database: 0

management:
  endpoints:
    web:
      exposure:
        include: health,info,prometheus,metrics
  metrics:
    export:
      prometheus:
        enabled: true

rate:
  limit:
    enabled: true
    history-days: 7
    safety-factor: 1.2
    smoothing-factor: 0.3
    default-limit: 100

server:
  port: 8080

五、性能对比

1. 测试场景

• 模拟正常流量：100 QPS
• 模拟高峰期流量：300 QPS
• 模拟低峰期流量：20 QPS
• 测试时间：24小时
• 监控频率：5分钟

2. 测试结果

3. 关键指标对比

六、最佳实践

1. 配置优化

• 合理设置历史数据天数：根据业务特点设置合适的历史数据天数
• 调整安全系数：根据系统重要性调整安全系数
• 优化平滑因子：根据流量波动程度调整平滑因子
• 设置合理的默认值：为新端点设置合理的默认限流阈值
• 定期清理历史数据：避免历史数据占用过多存储空间

2. 代码优化

• 异步处理：使用异步方式处理流量数据的采集和分析
• 缓存优化：合理使用缓存减少 Redis 访问
• 批量处理：对流量数据进行批量处理，减少网络开销
• 异常处理：妥善处理各种异常情况，确保系统稳定
• 监控覆盖：为限流相关的操作添加监控指标

3. 监控策略

• 实时监控：实时监控系统的流量和限流情况
• 历史分析：定期分析历史流量数据，优化限流策略
• 告警机制：当限流阈值频繁调整时，发送告警通知
• 可视化：提供限流情况的可视化面板
• A/B 测试：通过 A/B 测试验证不同限流策略的效果

4. 应急处理

• 手动覆盖：在特殊情况下允许手动覆盖限流阈值
• 降级策略：在系统压力过大时，实施服务降级策略
• 熔断机制：当系统出现异常时，自动熔断部分功能
• 流量预测：基于历史数据预测未来流量，提前调整限流阈值
• 事后分析：对限流事件进行事后分析，总结经验教训

七、总结与展望

方案总结

1. 实时监控：实时采集和分析系统的流量数据
2. 历史分析：基于历史流量数据进行统计和分析
3. 智能推荐：根据历史流量模式自动推荐合理的限流阈值
4. 动态调整：根据实时流量和历史数据动态调整限流阈值
5. 自适应学习：通过持续学习不断优化限流策略
6. 可扩展性：支持与各种监控系统和限流框架集成

未来优化方向

1. 机器学习：使用机器学习算法提高流量预测的准确性
2. 多维度分析：从更多维度（如用户类型、设备类型）分析流量
3. 分布式支持：支持分布式环境下的限流阈值调优
4. 实时预测：基于实时数据和历史模式进行流量预测
5. 智能告警：根据限流情况智能调整告警策略

技术价值

1. 提高系统稳定性：通过合理的限流，保护系统免受流量冲击
2. 提升用户体验：减少误杀，提高系统的可用性
3. 优化资源利用：根据实际流量调整限流阈值，提高资源利用率
4. 降低运维成本：减少人工干预，实现自动调优
5. 数据驱动决策：基于数据进行限流策略的优化

八、写在最后

限流阈值的设置是一个复杂的问题，固定的阈值无法适应业务的动态变化。通过基于历史流量的限流阈值动态调优方案，我们可以实现限流阈值的自动调整，提高系统的稳定性和可用性。

当然，这套方案也不是银弹，它有以下局限性：

• 数据依赖：依赖于历史流量数据，对于新系统可能效果不佳
• 计算开销：需要进行大量的数据分析和计算
• 配置复杂：需要根据实际情况进行合理配置
• 响应延迟：调整限流阈值需要一定的时间

但对于需要高可靠性的系统，这套方案已经足够解决问题，而且稳定可靠。

希望这篇文章能给你带来一些启发，帮助你在实际项目中更好地处理限流阈值的设置问题。

如果你在使用这套方案的过程中有其他经验或困惑，欢迎在评论区留言交流！

服务端技术精选，专注分享后端开发实战经验，让技术落地更简单。

如果你觉得这篇文章有用，欢迎点赞、在看、分享三连！