IT技术社区_专业开发者平台_最新编程教程与解决方案 -沐辰社区

Spring Cloud Gateway限流案例

wen java案例 3

本文目录导读：

Spring Cloud Gateway限流案例

  1. 项目依赖配置
  2. 基础配置文件
  3. 限流Key解析器配置
  4. 自定义限流过滤器
  5. 使用Bucket4j实现更复杂限流
  6. 动态限流配置
  7. 全局限流配置
  8. 高级配置示例
  9. 监控和指标
  10. 完整示例：统一限流管理
  11. 使用示例

我来为您提供一个完整的Spring Cloud Gateway限流案例，涵盖多种限流策略。

项目依赖配置

Maven依赖

<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-starter-gateway</artifactId>
</dependency>
<!-- Redis实现令牌桶限流 -->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-redis-reactive</artifactId>
</dependency>
<!-- 限流算法支持 -->
<dependency>
    <groupId>com.github.vladimir-bukhtoyarov</groupId>
    <artifactId>bucket4j-core</artifactId>
    <version>7.6.0</version>
</dependency>
<dependency>
    <groupId>com.github.vladimir-bukhtoyarov</groupId>
    <artifactId>bucket4j-redis</artifactId>
    <version>7.6.0</version>
</dependency>

基础配置文件

application.yml

spring:
  application:
    name: gateway-service
  redis:
    host: localhost
    port: 6379
    database: 0
  cloud:
    gateway:
      routes:
        - id: user-service
          uri: lb://user-service
          predicates:
            - Path=/api/user/**
          filters:
            - name: RequestRateLimiter
              args:
                redis-rate-limiter.replenishRate: 10
                redis-rate-limiter.burstCapacity: 20
                redis-rate-limiter.requestedTokens: 1
                key-resolver: "#{@userKeyResolver}"
        - id: order-service
          uri: lb://order-service
          predicates:
            - Path=/api/order/**
          filters:
            - name: RequestRateLimiter
              args:
                redis-rate-limiter.replenishRate: 5
                redis-rate-limiter.burstCapacity: 10
                redis-rate-limiter.requestedTokens: 1
                key-resolver: "#{@ipKeyResolver}"

限流Key解析器配置

KeyResolver配置类

@Configuration
public class RateLimiterConfig {
    /**
     * 基于用户ID的限流
     */
    @Bean
    public KeyResolver userKeyResolver() {
        return exchange -> {
            // 从需要中获取用户信息
            String userId = exchange.getRequest().getHeaders().getFirst("X-User-Id");
            if (userId == null || userId.isEmpty()) {
                // 默认使用IP作为标识
                userId = getClientIp(exchange);
            }
            return Mono.just(userId);
        };
    }
    /**
     * 基于IP地址的限流
     */
    @Bean
    public KeyResolver ipKeyResolver() {
        return exchange -> Mono.just(getClientIp(exchange));
    }
    /**
     * 基于API Key的限流
     */
    @Bean
    public KeyResolver apiKeyResolver() {
        return exchange -> {
            String apiKey = exchange.getRequest().getHeaders().getFirst("X-API-Key");
            if (apiKey == null || apiKey.isEmpty()) {
                return Mono.just("unknown_api_key");
            }
            return Mono.just(apiKey);
        };
    }
    private String getClientIp(ServerWebExchange exchange) {
        String ip = exchange.getRequest().getHeaders().getFirst("X-Forwarded-For");
        if (ip == null || ip.isEmpty()) {
            ip = exchange.getRequest().getRemoteAddress() != null ? 
                exchange.getRequest().getRemoteAddress().getAddress().getHostAddress() : "unknown";
        } else {
            ip = ip.split(",")[0].trim();
        }
        return ip;
    }
}

自定义限流过滤器

自定义限流实现

@Component
public class CustomRateLimiterFilter implements GlobalFilter, Ordered {
    private final ReactiveStringRedisTemplate redisTemplate;
    private final RateLimiterConfig rateLimiterConfig;
    @Autowired
    public CustomRateLimiterFilter(ReactiveStringRedisTemplate redisTemplate, 
                                   RateLimiterConfig rateLimiterConfig) {
        this.redisTemplate = redisTemplate;
        this.rateLimiterConfig = rateLimiterConfig;
    }
    @Override
    public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
        // 获取请求路径
        String path = exchange.getRequest().getURI().getPath();
        // 获取客户端IP
        String clientIp = exchange.getRequest().getRemoteAddress() != null ? 
            exchange.getRequest().getRemoteAddress().getAddress().getHostAddress() : "unknown";
        // 构建Redis key
        String rateLimitKey = "rate_limit:" + clientIp + ":" + path;
        // 实现简单的滑动窗口限流
        return isAllowed(rateLimitKey)
            .flatMap(allowed -> {
                if (allowed) {
                    return chain.filter(exchange);
                } else {
                    // 返回429状态码
                    exchange.getResponse().setStatusCode(HttpStatus.TOO_MANY_REQUESTS);
                    exchange.getResponse().getHeaders().setContentType(MediaType.APPLICATION_JSON);
                    return exchange.getResponse().writeWith(
                        Mono.just(exchange.getResponse().bufferFactory()
                            .wrap("{\"code\":429,\"message\":\"请求过于频繁\"}".getBytes()))
                    );
                }
            });
    }
    private Mono<Boolean> isAllowed(String key) {
        return redisTemplate.opsForValue()
            .increment(key)
            .flatMap(count -> {
                if (count == 1) {
                    // 设置过期时间，例如1分钟
                    return redisTemplate.expire(key, Duration.ofMinutes(1))
                        .map(expired -> true);
                }
                return Mono.just(count <= 100); // 每分钟最多100次请求
            });
    }
    @Override
    public int getOrder() {
        return -100; // 高优先级执行
    }
}

使用Bucket4j实现更复杂限流

Bucket4j配置类

@Configuration
public class Bucket4jConfig {
    private final RedisTemplate<String, Object> redisTemplate;
    @Autowired
    public Bucket4jConfig(RedisTemplate<String, Object> redisTemplate) {
        this.redisTemplate = redisTemplate;
    }
    /**
     * 创建基于IP的限流桶
     */
    @Bean
    public Bucket createIpBucket() {
        // 创建Bucket4j配置
        Bandwidth limit = Bandwidth.classic(10, Refill.greedy(10, Duration.ofMinutes(1)));
        // 使用Redis存储桶状态
        ProxyManager<String> buckets = Bucket4j.extension(Redis.class)
            .proxyManagerForMap(redisTemplate.opsForValue().getOperations().getConnectionFactory());
        return Bucket.builder()
            .addLimit(limit)
            .build();
    }
    /**
     * 获取或创建Bucket
     */
    public Bucket getOrCreateBucket(String key) {
        ProxyManager<String> buckets = Bucket4j.extension(Redis.class)
            .proxyManagerForMap(redisTemplate.opsForValue().getOperations().getConnectionFactory());
        Function<String, Bucket> bucketConfiguration = k -> Bucket.builder()
            .addLimit(Bandwidth.classic(10, Refill.greedy(10, Duration.ofMinutes(1))))
            .build();
        return buckets.getProxy(key, bucketConfiguration);
    }
}

使用Bucket4j的过滤器

@Component
public class Bucket4jRateLimiterFilter implements GatewayFilter {
    private final Bucket4jConfig bucket4jConfig;
    @Autowired
    public Bucket4jRateLimiterFilter(Bucket4jConfig bucket4jConfig) {
        this.bucket4jConfig = bucket4jConfig;
    }
    @Override
    public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
        // 获取客户端IP作为key
        String clientIp = exchange.getRequest().getRemoteAddress() != null ? 
            exchange.getRequest().getRemoteAddress().getAddress().getHostAddress() : "unknown";
        // 获取或创建Bucket
        Bucket bucket = bucket4jConfig.getOrCreateBucket(clientIp);
        // 尝试获取令牌
        ConsumptionProbe probe = bucket.tryConsumeAndReturnRemaining(1);
        if (probe.isConsumed()) {
            // 添加剩余令牌数到响应头
            exchange.getResponse().getHeaders().add("X-Rate-Limit-Remaining", 
                String.valueOf(probe.getRemainingTokens()));
            return chain.filter(exchange);
        } else {
            // 计算重试时间
            long waitForRefill = probe.getNanosToWaitForRefill() / 1_000_000_000;
            exchange.getResponse().getHeaders().add("X-Rate-Limit-Retry-After-Seconds", 
                String.valueOf(waitForRefill));
            // 返回429状态码
            exchange.getResponse().setStatusCode(HttpStatus.TOO_MANY_REQUESTS);
            return exchange.getResponse().setComplete();
        }
    }
}

动态限流配置

限流规则管理类

@Service
public class RateLimitRuleService {
    private final RedisTemplate<String, Object> redisTemplate;
    private final Map<String, RateLimitRule> ruleCache = new ConcurrentHashMap<>();
    @Autowired
    public RateLimitRuleService(RedisTemplate<String, Object> redisTemplate) {
        this.redisTemplate = redisTemplate;
        loadInitialRules();
    }
    /**
     * 更新限流规则
     */
    public void updateRule(String key, RateLimitRule rule) {
        redisTemplate.opsForValue().set("rate_rule:" + key, rule);
        ruleCache.put(key, rule);
    }
    /**
     * 获取规则
     */
    public RateLimitRule getRule(String key) {
        // 先从缓存获取
        RateLimitRule rule = ruleCache.get(key);
        if (rule == null) {
            // 从Redis获取
            rule = (RateLimitRule) redisTemplate.opsForValue().get("rate_rule:" + key);
            if (rule != null) {
                ruleCache.put(key, rule);
            }
        }
        return rule;
    }
    /**
     * 动态过滤器
     */
    public Mono<Void> applyDynamicRateLimit(ServerWebExchange exchange, GatewayFilterChain chain) {
        String clientIp = exchange.getRequest().getRemoteAddress() != null ? 
            exchange.getRequest().getRemoteAddress().getAddress().getHostAddress() : "unknown";
        RateLimitRule rule = getRule(clientIp);
        if (rule == null) {
            // 使用默认规则
            rule = new RateLimitRule(10, 1, 60); // 每秒10个请求，突发1个，窗口1分钟
        }
        // 应用动态规则
        return isAllowedWithRule(clientIp, rule)
            .flatMap(allowed -> {
                if (allowed) {
                    return chain.filter(exchange);
                } else {
                    exchange.getResponse().setStatusCode(HttpStatus.TOO_MANY_REQUESTS);
                    return exchange.getResponse().setComplete();
                }
            });
    }
    private Mono<Boolean> isAllowedWithRule(String key, RateLimitRule rule) {
        // 使用Lua脚本实现原子性操作
        String script = """
            local key = KEYS[1]
            local limit = tonumber(ARGV[1])
            local window = tonumber(ARGV[2])
            local current = redis.call('INCR', key)
            if current == 1 then
                redis.call('EXPIRE', key, window)
            end
            return current <= limit
            """;
        return redisTemplate.execute(
            new DefaultRedisScript<>(script, Boolean.class),
            Arrays.asList("rate:" + key),
            String.valueOf(rule.getLimit()),
            String.valueOf(rule.getWindowSeconds())
        );
    }
}

全局限流配置

全局过滤器配置

@Configuration
public class GatewayFilterConfig {
    @Bean
    public GlobalFilter customGlobalFilter() {
        return (exchange, chain) -> {
            // 检查是否是API请求
            if (exchange.getRequest().getURI().getPath().startsWith("/api/")) {
                // 添加自定义头
                exchange.getRequest().mutate()
                    .header("X-Gateway-Request-Time", String.valueOf(System.currentTimeMillis()))
                    .build();
            }
            return chain.filter(exchange);
        };
    }
    @Bean
    public GlobalFilter rateLimitGlobalFilter(RateLimitRuleService rateLimitRuleService) {
        return (exchange, chain) -> rateLimitRuleService
            .applyDynamicRateLimit(exchange, chain);
    }
}

高级配置示例

多级限流配置

spring:
  cloud:
    gateway:
      routes:
        - id: api-routes
          uri: lb://api-service
          predicates:
            - Path=/api/**
          filters:
            # 第一级：限速
            - RequestRateLimiter=5, 10
            # 第二级：限流
            - name: RequestRateLimiter
              args:
                redis-rate-limiter.replenishRate: 5
                redis-rate-limiter.burstCapacity: 10
                key-resolver: "#{@userKeyResolver}"

自定义错误处理

@Configuration
public class RateLimitErrorHandler {
    @Bean
    public ErrorWebExceptionHandler rateLimitExceptionHandler() {
        return (exchange, ex) -> {
            if (ex instanceof RateLimitExceededException) {
                exchange.getResponse().setStatusCode(HttpStatus.TOO_MANY_REQUESTS);
                exchange.getResponse().getHeaders().setContentType(MediaType.APPLICATION_JSON);
                String errorMessage = """
                    {
                        "code": 429,
                        "message": "请求过于频繁，请稍后再试",
                        "retryAfter": 60,
                        "timestamp": "%s"
                    }
                    """.formatted(LocalDateTime.now());
                return exchange.getResponse().writeWith(
                    Mono.just(exchange.getResponse().bufferFactory().wrap(errorMessage.getBytes()))
                );
            }
            return Mono.error(ex);
        };
    }
}

监控和指标

监控配置

@Component
public class RateLimitMetrics {
    private final MeterRegistry meterRegistry;
    @Autowired
    public RateLimitMetrics(MeterRegistry meterRegistry) {
        this.meterRegistry = meterRegistry;
    }
    public void recordRateLimit(String key, boolean allowed) {
        meterRegistry.counter("gateway.rate.limit.requests",
            "key", key,
            "result", allowed ? "allowed" : "blocked"
        ).increment();
    }
    public void recordLatency(String key, long latency) {
        meterRegistry.timer("gateway.rate.limit.latency",
            "key", key
        ).record(Duration.ofMillis(latency));
    }
}

完整示例：统一限流管理

限流管理器

@Service
public class UnifiedRateLimiter {
    @Autowired
    private ReactiveStringRedisTemplate redisTemplate;
    /**
     * 统一的限流入口
     */
    public Mono<RateLimitResult> checkRateLimit(String key, RateLimitType type) {
        return switch (type) {
            case IP -> checkIpRateLimit(key);
            case USER -> checkUserRateLimit(key);
            case API -> checkApiRateLimit(key);
            case CUSTOM -> checkCustomRateLimit(key);
        };
    }
    private Mono<RateLimitResult> checkIpRateLimit(String ip) {
        // IP限流：每秒10次，突发20次
        return checkLimit("ip:" + ip, 10, 20, 1);
    }
    private Mono<RateLimitResult> checkUserRateLimit(String userId) {
        // 用户限流：每分钟100次
        return checkLimit("user:" + userId, 100, 200, 60);
    }
    private Mono<RateLimitResult> checkApiRateLimit(String apiKey) {
        // API限流：每小时1000次
        return checkLimit("api:" + apiKey, 1000, 2000, 3600);
    }
    private Mono<RateLimitResult> checkCustomRateLimit(String key) {
        // 自定义限流
        return checkLimit("custom:" + key, 50, 100, 60);
    }
    private Mono<RateLimitResult> checkLimit(String key, int limit, int burst, int windowSeconds) {
        String script = """
            local key = KEYS[1]
            local limit = tonumber(ARGV[1])
            local burst = tonumber(ARGV[2])
            local window = tonumber(ARGV[3])
            local current = redis.call('INCR', key)
            if current == 1 then
                redis.call('EXPIRE', key, window)
            end
            local maxAllowed = (current <= burst) and limit or (limit + (burst - limit) * (window - redis.call('TTL', key)) / window)
            if current <= maxAllowed then
                return {1, current}
            else
                return {0, current}
            end
            """;
        return redisTemplate.execute(
            new DefaultRedisScript<>(script, List.class),
            Arrays.asList(key),
            String.valueOf(limit),
            String.valueOf(burst),
            String.valueOf(windowSeconds)
        ).map(result -> {
            List<Long> values = (List<Long>) result;
            return new RateLimitResult(values.get(0) == 1, 
                                      values.get(1), 
                                      limit, 
                                      windowSeconds);
        });
    }
    public record RateLimitResult(boolean allowed, long currentCount, long limit, int windowSeconds) {}
}
// 限流类型枚举
public enum RateLimitType {
    IP, USER, API, CUSTOM
}

使用示例

在业务逻辑中使用

@RestController
@RequestMapping("/api")
public class TestController {
    @Autowired
    private UnifiedRateLimiter rateLimiter;
    @GetMapping("/test")
    public Mono<ResponseEntity<?>> test() {
        return rateLimiter.checkRateLimit("test-key", RateLimitType.CUSTOM)
            .flatMap(result -> {
                if (result.allowed()) {
                    return Mono.just(ResponseEntity.ok("请求成功"));
                } else {
                    return Mono.just(
                        ResponseEntity.status(429)
                            .header("X-RateLimit-Remaining", 
                                String.valueOf(result.limit() - result.currentCount()))
                            .body("请求过于频繁")
                    );
                }
            });
    }
}

这个完整的案例提供了多种限流方案,您可以根据实际需求选择合适的实现方式，主要特性包括：

  1. 基于Redis的令牌桶限流
  2. 基于IP、用户、API Key的多维度限流
  3. 动态规则配置
  4. 全局和局部限流结合
  5. 监控和指标收集
  6. 灵活的错误处理机制

上一篇Spring Cloud Security案例

下一篇Spring Cloud OAuth2认证案例

相关文章

抱歉，评论功能暂时关闭!