网关与限流熔断

客户端需要知道所有服务地址:
┌─────────┐
│ 客户端   │
└─────────┘
    ↓ 直接调用多个服务
    ├─→ 用户服务 (http://user-service:8081)
    ├─→ 订单服务 (http://order-service:8082)
    ├─→ 商品服务 (http://product-service:8083)
    └─→ 支付服务 (http://payment-service:8084)

                    ┌─────────────┐
                    │   客户端     │
                    └─────────────┘
                          ↓
                    ┌─────────────┐
                    │  API 网关    │ ← 横切能力集中治理
                    │ - 路由转发   │
                    │ - 认证授权   │
                    │ - 限流熔断   │
                    │ - 灰度发布   │
                    │ - 日志监控   │
                    └─────────────┘
                          ↓
        ┌─────────────────┼─────────────────┐
        ↓                 ↓                 ↓
  ┌──────────┐      ┌──────────┐      ┌──────────┐
  │ 用户服务  │      │ 订单服务  │      │ 商品服务  │
  └──────────┘      └──────────┘      └──────────┘

// Spring Cloud Gateway 路由配置
@Bean
public RouteLocator customRouteLocator(RouteLocatorBuilder builder) {
    return builder.routes()
        .route("user-service", r -> r.path("/api/users/**")
            .filters(f -> f.stripPrefix(1))
            .uri("lb://user-service"))
        .route("order-service", r -> r.path("/api/orders/**")
            .filters(f -> f.stripPrefix(1))
            .uri("lb://order-service"))
        .build();
}

// 网关鉴权过滤器
@Component
public class AuthFilter implements GlobalFilter, Ordered {
    
    @Override
    public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
        String token = exchange.getRequest().getHeaders().getFirst("Authorization");
        
        if (token == null || !validateToken(token)) {
            exchange.getResponse().setStatusCode(HttpStatus.UNAUTHORIZED);
            return exchange.getResponse().setComplete();
        }
        
        // 解析用户信息并透传
        UserInfo user = parseToken(token);
        ServerHttpRequest request = exchange.getRequest().mutate()
            .header("X-User-Id", user.getId())
            .header("X-User-Name", user.getName())
            .build();
        
        return chain.filter(exchange.mutate().request(request).build());
    }
    
    @Override
    public int getOrder() {
        return -100;
    }
}

// 基于 Redis 的限流配置
@Bean
public RedisRateLimiter redisRateLimiter() {
    return new RedisRateLimiter(100, 200); // replenishRate=100, burstCapacity=200
}

@Bean
public RouteLocator customRouteLocator(RouteLocatorBuilder builder) {
    return builder.routes()
        .route("user-service", r -> r.path("/api/users/**")
            .filters(f -> f
                .stripPrefix(1)
                .requestRateLimiter(c -> c
                    .setRateLimiter(redisRateLimiter())
                    .setKeyResolver(ipKeyResolver())
                )
            )
            .uri("lb://user-service"))
        .build();
}

@Bean
public KeyResolver ipKeyResolver() {
    return exchange -> Mono.just(
        exchange.getRequest().getRemoteAddress().getAddress().getHostAddress()
    );
}

# application.yml
spring:
  cloud:
    gateway:
      routes:
        - id: user-service
          uri: lb://user-service
          predicates:
            - Path=/api/users/**
          filters:
            - name: CircuitBreaker
              args:
                name: userServiceCircuitBreaker
                fallbackUri: forward:/fallback/user-service

// 灰度路由配置(示意)
@Component
public class GrayLoadBalancer implements ReactorServiceInstanceLoadBalancer {

    private final ServiceInstanceListSupplier supplier;

    public GrayLoadBalancer(ServiceInstanceListSupplier supplier, String serviceId) {
        this.supplier = supplier;
    }

    @Override
    public Mono<Response<ServiceInstance>> choose(Request request) {
        // Spring Cloud LoadBalancer 调用网关场景下,context 即 ServerWebExchange
        ServerWebExchange exchange = (ServerWebExchange) request.getContext();
        // 从请求头获取灰度标识
        String grayVersion = exchange.getRequest().getHeaders().getFirst("X-Gray-Version");

        // 选择对应版本的实例
        return supplier.get().next()
            .map(instances -> new DefaultResponse(selectInstance(instances, grayVersion)));
    }

    private ServiceInstance selectInstance(List<ServiceInstance> instances, String grayVersion) {
        // 按 metadata 中的 version 标记筛选,未命中则退回普通实例
        return instances.stream()
            .filter(i -> grayVersion == null || grayVersion.equals(i.getMetadata().get("version")))
            .findFirst()
            .orElse(instances.get(0));
    }
}

// 全局日志过滤器
@Component
public class AccessLogFilter implements GlobalFilter, Ordered {
    
    private static final Logger log = LoggerFactory.getLogger(AccessLogFilter.class);
    
    @Override
    public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
        long startTime = System.currentTimeMillis();
        
        return chain.filter(exchange).then(Mono.fromRunnable(() -> {
            long duration = System.currentTimeMillis() - startTime;
            
            log.info("[网关访问日志] method={}, path={}, status={}, duration={}ms, ip={}",
                exchange.getRequest().getMethod(),
                exchange.getRequest().getPath(),
                exchange.getResponse().getStatusCode(),
                duration,
                exchange.getRequest().getRemoteAddress()
            );
        }));
    }
    
    @Override
    public int getOrder() {
        return -200;
    }
}

@Bean
public RouteLocator customRouteLocator(RouteLocatorBuilder builder) {
    return builder.routes()
        .route("grpc-proxy", r -> r.path("/api/grpc/**")
            .filters(f -> f.stripPrefix(2))
            .uri("http://grpc-proxy:50051"))
        .build();
}

                    ┌─────────────┐
                    │   客户端     │
                    └─────────────┘
                          ↓
                    ┌─────────────┐
                    │ API 网关集群 │ ← 所有流量经过统一网关
                    └─────────────┘
                          ↓
        ┌─────────────────┼─────────────────┐
        ↓                 ↓                 ↓
    服务实例          服务实例          服务实例

                    ┌─────────────┐
                    │   客户端     │
                    └─────────────┘
                          ↓
        ┌─────────────────┼─────────────────┐
        ↓                 ↓                 ↓
  ┌──────────┐      ┌──────────┐      ┌──────────┐
  │ Sidecar  │      │ Sidecar  │      │ Sidecar  │ ← 每个服务旁边都有网关代理
  │ Gateway  │      │ Gateway  │      │ Gateway  │
  └──────────┘      └──────────┘      └──────────┘
        ↓                 ↓                 ↓
  ┌──────────┐      ┌──────────┐      ┌──────────┐
  │ 用户服务  │      │ 订单服务  │      │ 商品服务  │
  └──────────┘      └──────────┘      └──────────┘

// 响应式过滤器
@Component
public class ReactiveAuthFilter implements GlobalFilter {
    
    @Override
    public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
        return Mono.fromCallable(() -> validateToken(exchange))
            .flatMap(valid -> {
                if (valid) {
                    return chain.filter(exchange);
                } else {
                    exchange.getResponse().setStatusCode(HttpStatus.UNAUTHORIZED);
                    return exchange.getResponse().setComplete();
                }
            });
    }
}

// 响应缓存
@Component
public class CacheFilter implements GlobalFilter {
    
    @Autowired
    private RedisTemplate<String, String> redisTemplate;
    
    @Override
    public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
        String cacheKey = generateCacheKey(exchange.getRequest());
        
        // 尝试从缓存获取
        return Mono.fromCallable(() -> redisTemplate.opsForValue().get(cacheKey))
            .flatMap(cached -> {
                if (cached != null) {
                    // 返回缓存内容
                    exchange.getResponse().getHeaders().setContentType(MediaType.APPLICATION_JSON);
                    DataBuffer buffer = exchange.getResponse().bufferFactory()
                        .wrap(cached.getBytes(StandardCharsets.UTF_8));
                    return exchange.getResponse().writeWith(Mono.just(buffer));
                } else {
                    // 执行请求并缓存结果
                    return cacheResponse(exchange, chain, cacheKey);
                }
            });
    }
}

# application.yml
spring:
  cloud:
    gateway:
      httpclient:
        pool:
          type: ELASTIC
          max-idle-time: 15000
          max-connections: 500
        connect-timeout: 3000
        response-timeout: 5000

// 内置断言工厂
@Bean
public RouteLocator customRouteLocator(RouteLocatorBuilder builder) {
    return builder.routes()
        // 路径匹配
        .route("path-route", r -> r.path("/api/users/**")
            .uri("lb://user-service"))
        
        // 方法匹配
        .route("method-route", r -> r.method("GET")
            .uri("lb://user-service"))
        
        // 请求头匹配
        .route("header-route", r -> r.header("X-Request-Id", "\\d+")
            .uri("lb://user-service"))
        
        // 时间匹配
        .route("time-route", r -> r.between(
            LocalDateTime.of(2024, 1, 1, 0, 0),
            LocalDateTime.of(2024, 12, 31, 23, 59))
            .uri("lb://user-service"))
        
        // Cookie 匹配
        .route("cookie-route", r -> r.cookie("session", "^[a-z]+$")
            .uri("lb://user-service"))
        
        .build();
}

// 内置过滤器
@Bean
public RouteLocator customRouteLocator(RouteLocatorBuilder builder) {
    return builder.routes()
        .route("user-service", r -> r.path("/api/users/**")
            .filters(f -> f
                // 添加请求头
                .addRequestHeader("X-Request-Foo", "Bar")
                
                // 添加请求参数
                .addRequestParameter("foo", "bar")
                
                // 添加响应头
                .addResponseHeader("X-Response-Foo", "Bar")
                
                // 去掉路径前缀
                .stripPrefix(1)
                
                // 添加路径前缀
                .prefixPath("/api")
                
                // 重定向
                .redirect(HttpStatus.MOVED_PERMANENTLY, "https://example.com")
                
                // 重试
                .retry(config -> config
                    .setRetries(3)
                    .setStatuses(HttpStatus.INTERNAL_SERVER_ERROR)
                    .setMethods(HttpMethod.GET)
                )
            )
            .uri("lb://user-service"))
        .build();
}

# application.yml
spring:
  cloud:
    gateway:
      # 全局跨域配置
      globalcors:
        cors-configurations:
          '[/**]':
            # 注意:allowCredentials=true 时 allowedOrigins 不能为 "*",需改用 allowedOriginPatterns
            allowedOriginPatterns: "*"
            allowedMethods:
              - GET
              - POST
              - PUT
              - DELETE
            allowedHeaders: "*"
            allowCredentials: true
            maxAge: 3600
      
      # 路由配置
      routes:
        # 用户服务
        - id: user-service
          uri: lb://user-service
          predicates:
            - Path=/api/users/**
          filters:
            - StripPrefix=1
            - name: RequestRateLimiter
              args:
                redis-rate-limiter.replenishRate: 100
                redis-rate-limiter.burstCapacity: 200
                key-resolver: "#{@ipKeyResolver}"
        
        # 订单服务
        - id: order-service
          uri: lb://order-service
          predicates:
            - Path=/api/orders/**
          filters:
            - StripPrefix=1
            - name: CircuitBreaker
              args:
                name: orderServiceCircuitBreaker
                fallbackUri: forward:/fallback/order-service
        
        # 商品服务
        - id: product-service
          uri: lb://product-service
          predicates:
            - Path=/api/products/**
            - Method=GET
          filters:
            - StripPrefix=1
            - AddRequestHeader=X-Request-Source, gateway
      
      # 默认过滤器
      default-filters:
        - AddResponseHeader=X-Response-Source, gateway
      
      # 全局配置
      httpclient:
        connect-timeout: 3000
        response-timeout: 5000
        pool:
          type: elastic
          max-idle-time: 15000

@Component
public class AuthGlobalFilter implements GlobalFilter, Ordered {
    
    private static final Logger log = LoggerFactory.getLogger(AuthGlobalFilter.class);
    
    @Override
    public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
        String token = exchange.getRequest().getHeaders().getFirst("Authorization");
        
        if (token == null) {
            log.warn("请求缺少认证token, path={}", exchange.getRequest().getPath());
            exchange.getResponse().setStatusCode(HttpStatus.UNAUTHORIZED);
            return exchange.getResponse().setComplete();
        }
        
        try {
            // 验证 token
            UserInfo user = JwtUtil.parseToken(token);
            
            // 透传用户信息
            ServerHttpRequest request = exchange.getRequest().mutate()
                .header("X-User-Id", user.getId().toString())
                .header("X-User-Name", user.getUsername())
                .header("X-User-Roles", String.join(",", user.getRoles()))
                .build();
            
            return chain.filter(exchange.mutate().request(request).build());
            
        } catch (Exception e) {
            log.error("Token验证失败: {}", e.getMessage());
            exchange.getResponse().setStatusCode(HttpStatus.UNAUTHORIZED);
            return exchange.getResponse().setComplete();
        }
    }
    
    @Override
    public int getOrder() {
        return -100; // 优先级高
    }
}

@Component
public class RequestTimeGatewayFilterFactory 
        extends AbstractGatewayFilterFactory<RequestTimeGatewayFilterFactory.Config> {
    
    public RequestTimeGatewayFilterFactory() {
        super(Config.class);
    }
    
    @Override
    public GatewayFilter apply(Config config) {
        return (exchange, chain) -> {
            long startTime = System.currentTimeMillis();
            
            return chain.filter(exchange).then(Mono.fromRunnable(() -> {
                long duration = System.currentTimeMillis() - startTime;
                
                if (config.isShowHeader()) {
                    exchange.getResponse().getHeaders()
                        .add("X-Response-Time", duration + "ms");
                }
                
                log.info("请求耗时: {}ms, path={}", 
                    duration, 
                    exchange.getRequest().getPath()
                );
            }));
        };
    }
    
    public static class Config {
        private boolean showHeader = true;
        
        public boolean isShowHeader() {
            return showHeader;
        }
        
        public void setShowHeader(boolean showHeader) {
            this.showHeader = showHeader;
        }
    }
}

┌─────────────────────────────────────────────────────────┐
│                    Kong Gateway                          │
│  ┌────────────────────────────────────────────────────┐ │
│  │                 Kong Core                           │ │
│  │  ┌──────────┐  ┌──────────┐  ┌──────────┐        │ │
│  │  │ Router   │  │ Plugin   │  │ Admin    │        │ │
│  │  │ Engine   │  │ System   │  │ API      │        │ │
│  │  └──────────┘  └──────────┘  └──────────┘        │ │
│  └────────────────────────────────────────────────────┘ │
│  ┌────────────────────────────────────────────────────┐ │
│  │              Data Store (PostgreSQL)               │ │
│  └────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────┘

# 创建服务
curl -i -X POST http://localhost:8001/services \
  -d "name=user-service" \
  -d "url=http://user-service:8080"

# 创建路由
curl -i -X POST http://localhost:8001/services/user-service/routes \
  -d "name=user-route" \
  -d "paths[]=/api/users"

# 添加限流插件
curl -i -X POST http://localhost:8001/services/user-service/plugins \
  -d "name=rate-limiting" \
  -d "config.minute=100" \
  -d "config.policy=local"

┌─────────────────────────────────────────────────────────┐
│                    APISIX Gateway                        │
│  ┌────────────────────────────────────────────────────┐ │
│  │               APISIX Core                           │ │
│  │  ┌──────────┐  ┌──────────┐  ┌──────────┐        │ │
│  │  │ Router   │  │ Plugin   │  │ Admin    │        │ │
│  │  │ Matcher  │  │ Runtime  │  │ API      │        │ │
│  │  └──────────┘  └──────────┘  └──────────┘        │ │
│  └────────────────────────────────────────────────────┘ │
│  ┌────────────────────────────────────────────────────┐ │
│  │          etcd (配置存储)                            │ │
│  └────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────┘

# 创建路由
curl http://127.0.0.1:9180/apisix/admin/routes/1 \
-H 'X-API-KEY: edd1c9f034335f136f87ad84b625c8f1' -X PUT -d '
{
    "uri": "/api/users/*",
    "plugins": {
        "limit-req": {
            "rate": 100,
            "burst": 200,
            "rejected_code": 429
        },
        "jwt-auth": {}
    },
    "upstream": {
        "type": "roundrobin",
        "nodes": {
            "user-service:8080": 1
        }
    }
}'

首选: Spring Cloud Gateway
理由:
- 与 Spring Cloud 生态无缝集成
- 响应式编程,性能好
- 功能丰富,易于扩展
- 学习曲线低

首选: APISIX 或 Kong
理由:
- 云原生设计
- 支持动态配置
- 与 Kubernetes 深度集成
- 高性能

首选: Kong 或 Nginx
理由:
- 基于 Nginx,极致性能
- 成熟稳定
- 生产验证充分

首选: Envoy
理由:
- 服务网格标配
- 可观测性强
- 动态配置

正常情况:
请求量 < 系统容量 → 系统正常运行

限流场景:
请求量 > 系统容量 → 拒绝部分请求 → 系统仍能处理部分请求

无限流:
请求暴增 → 系统过载 → 所有请求失败 → 雪崩

有限流:
请求暴增 → 限流保护 → 拒绝多余请求 → 部分请求成功 → 系统稳定

无限流:
恶意流量占用所有资源 → 正常用户无法访问

有限流:
恶意流量被限制 → 正常用户可以正常访问

时间窗口: 1秒
限流阈值: 100次

|------- 窗口1 -------|------- 窗口2 -------|
0ms               1000ms              2000ms
请求数: 100          请求数: 50
√ 放行              √ 放行

public class FixedWindowRateLimiter {
    
    private final int limit; // 窗口内最大请求数
    private final long windowSize; // 窗口大小(毫秒)
    
    private AtomicInteger counter = new AtomicInteger(0);
    private AtomicLong lastWindowStart = new AtomicLong(System.currentTimeMillis());
    
    public FixedWindowRateLimiter(int limit, long windowSize) {
        this.limit = limit;
        this.windowSize = windowSize;
    }
    
    public boolean tryAcquire() {
        long currentTime = System.currentTimeMillis();
        long windowStart = lastWindowStart.get();
        
        // 检查是否需要重置窗口
        if (currentTime - windowStart >= windowSize) {
            synchronized (this) {
                windowStart = lastWindowStart.get();
                if (currentTime - windowStart >= windowSize) {
                    counter.set(0);
                    lastWindowStart.set(currentTime);
                }
            }
        }
        
        // 检查是否超过限制
        return counter.incrementAndGet() <= limit;
    }
}

|------- 窗口1 -------|------- 窗口2 -------|
                   900ms 100ms
                   请求100次 + 请求100次
                   
窗口边界处可能瞬间通过200次请求!

当前时间: 1500ms
窗口大小: 1000ms
统计范围: 500ms - 1500ms

|--- 窗口1 ---|--- 窗口2 ---|--- 窗口3 ---|
0ms        1000ms       2000ms       3000ms
          ↑
        当前统计窗口: 500ms - 1500ms

public class SlidingWindowRateLimiter {
    
    private final int limit; // 窗口内最大请求数
    private final long windowSize; // 窗口大小(毫秒)
    
    private final ConcurrentLinkedQueue<Long> timestamps = new ConcurrentLinkedQueue<>();
    
    public SlidingWindowRateLimiter(int limit, long windowSize) {
        this.limit = limit;
        this.windowSize = windowSize;
    }
    
    public boolean tryAcquire() {
        long currentTime = System.currentTimeMillis();
        long windowStart = currentTime - windowSize;
        
        // 移除窗口外的请求
        while (!timestamps.isEmpty() && timestamps.peek() < windowStart) {
            timestamps.poll();
        }
        
        // 检查是否超过限制
        if (timestamps.size() < limit) {
            timestamps.offer(currentTime);
            return true;
        }
        
        return false;
    }
}

令牌桶:
┌─────────────────────────┐
│  ┌───┐ ┌───┐ ┌───┐     │ ← 令牌(最大容量)
│  │ ● │ │ ● │ │   │ ... │
│  └───┘ └───┘ └───┘     │
└─────────────────────────┘
        ↑
    按速率放令牌(如100个/秒)

请求到来时:
- 有令牌 → 取走令牌 → 放行
- 无令牌 → 拒绝

public class TokenBucketRateLimiter {
    
    private final long capacity; // 桶容量
    private final long rate; // 放令牌速率(个/秒)
    
    private final AtomicLong tokens = new AtomicLong(0);
    private final AtomicLong lastRefillTime = new AtomicLong(System.currentTimeMillis());
    
    public TokenBucketRateLimiter(long capacity, long rate) {
        this.capacity = capacity;
        this.rate = rate;
    }
    
    public boolean tryAcquire() {
        return tryAcquire(1);
    }
    
    public boolean tryAcquire(long permits) {
        refill();
        
        while (true) {
            long currentTokens = tokens.get();
            if (currentTokens < permits) {
                return false;
            }
            
            if (tokens.compareAndSet(currentTokens, currentTokens - permits)) {
                return true;
            }
        }
    }
    
    private void refill() {
        long currentTime = System.currentTimeMillis();
        long lastTime = lastRefillTime.get();
        
        if (currentTime > lastTime) {
            long elapsed = currentTime - lastTime;
            long newTokens = (elapsed * rate) / 1000;
            
            if (newTokens > 0) {
                while (true) {
                    long currentTokens = tokens.get();
                    long newTokenCount = Math.min(currentTokens + newTokens, capacity);
                    
                    if (tokens.compareAndSet(currentTokens, newTokenCount)) {
                        lastRefillTime.compareAndSet(lastTime, currentTime);
                        break;
                    }
                }
            }
        }
    }
}

漏桶:
请求 → ┌──────────────┐ → 处理
       │   ┌──────┐   │
       │   │ ●●●● │   │ ← 请求排队
       │   └──────┘   │
       └──────────────┘
              ↓
         固定速率流出
         (如100个/秒)

public class LeakyBucketRateLimiter {
    
    private final long capacity; // 桶容量
    private final long rate; // 流出速率(个/秒)
    
    private final ConcurrentLinkedQueue<Long> queue = new ConcurrentLinkedQueue<>();
    private final AtomicLong lastLeakTime = new AtomicLong(System.currentTimeMillis());
    
    public LeakyBucketRateLimiter(long capacity, long rate) {
        this.capacity = capacity;
        this.rate = rate;
    }
    
    public boolean tryAcquire() {
        leak();
        
        if (queue.size() < capacity) {
            queue.offer(System.currentTimeMillis());
            return true;
        }
        
        return false;
    }
    
    private void leak() {
        long currentTime = System.currentTimeMillis();
        long lastTime = lastLeakTime.get();
        
        if (currentTime > lastTime) {
            long elapsed = currentTime - lastTime;
            long leaks = (elapsed * rate) / 1000;
            
            if (leaks > 0) {
                // 移除已处理的请求
                for (long i = 0; i < leaks && !queue.isEmpty(); i++) {
                    queue.poll();
                }
                
                lastLeakTime.compareAndSet(lastTime, currentTime);
            }
        }
    }
}

// 按用户ID限流
@Bean
public KeyResolver userKeyResolver() {
    return exchange -> Mono.just(
        exchange.getRequest().getHeaders().getFirst("X-User-Id")
    );
}

// 配置
spring:
  cloud:
    gateway:
      routes:
        - id: user-service
          uri: lb://user-service
          predicates:
            - Path=/api/users/**
          filters:
            - name: RequestRateLimiter
              args:
                redis-rate-limiter.replenishRate: 10
                redis-rate-limiter.burstCapacity: 20
                key-resolver: "#{@userKeyResolver}"

// 按接口路径限流
@Bean
public KeyResolver apiKeyResolver() {
    return exchange -> Mono.just(
        exchange.getRequest().getPath().value()
    );
}

// 按IP限流
@Bean
public KeyResolver ipKeyResolver() {
    return exchange -> Mono.just(
        exchange.getRequest().getRemoteAddress().getAddress().getHostAddress()
    );
}

// 按租户限流
@Bean
public KeyResolver tenantKeyResolver() {
    return exchange -> Mono.just(
        exchange.getRequest().getHeaders().getFirst("X-Tenant-Id")
    );
}

// 组合限流:用户+接口
@Bean
public KeyResolver compositeKeyResolver() {
    return exchange -> Mono.just(
        exchange.getRequest().getHeaders().getFirst("X-User-Id") + ":" +
        exchange.getRequest().getPath().value()
    );
}

集群环境:
服务实例1: 限流100/秒
服务实例2: 限流100/秒
服务实例3: 限流100/秒

总限流: 300/秒

问题:
- 负载不均时,某个实例可能先被打满
- 无法实现全局限流

@Component
public class RedisRateLimiter {
    
    @Autowired
    private StringRedisTemplate redisTemplate;
    
    private static final String SCRIPT =
        "local key = KEYS[1]\n" +
        "local limit = tonumber(ARGV[1])\n" +
        "local window = tonumber(ARGV[2])\n" +
        "local current = tonumber(ARGV[3])\n" +
        "\n" +
        "redis.call('zremrangebyscore', key, 0, current - window * 1000)\n" +
        "local count = redis.call('zcard', key)\n" +
        "\n" +
        "if count < limit then\n" +
        "    redis.call('zadd', key, current, current .. '-' .. math.random())\n" +
        "    redis.call('expire', key, window)\n" +
        "    return 1\n" +
        "else\n" +
        "    return 0\n" +
        "end";
    
    public boolean tryAcquire(String key, int limit, int windowSeconds) {
        Long result = redisTemplate.execute(
            new DefaultRedisScript<>(SCRIPT, Long.class),
            Collections.singletonList(key),
            String.valueOf(limit),
            String.valueOf(windowSeconds),
            String.valueOf(System.currentTimeMillis())
        );
        
        return result != null && result == 1;
    }
}

spring:
  cloud:
    gateway:
      routes:
        - id: user-service
          uri: lb://user-service
          predicates:
            - Path=/api/users/**
          filters:
            - name: RequestRateLimiter
              args:
                redis-rate-limiter.replenishRate: 100
                redis-rate-limiter.burstCapacity: 200
                key-resolver: "#{@ipKeyResolver}"
  
  redis:
    host: localhost
    port: 6379

# 核心交易接口 - 高限流阈值
核心接口:
  - /api/orders/create: 1000/秒
  - /api/payments/process: 500/秒

# 非核心查询接口 - 中限流阈值
非核心接口:
  - /api/products/list: 500/秒
  - /api/users/profile: 200/秒

# 边缘接口 - 低限流阈值
边缘接口:
  - /api/comments/list: 100/秒
  - /api/recommendations: 50/秒

# VIP用户 - 高限额
VIP用户:
  - 查询: 200/秒
  - 下单: 50/秒

# 普通用户 - 中等限额
普通用户:
  - 查询: 100/秒
  - 下单: 10/秒

# 匿名用户 - 低限额
匿名用户:
  - 查询: 20/秒
  - 下单: 禁止

@Configuration
@RefreshScope
public class RateLimitConfig {
    
    @Value("${rate.limit.user-query:100}")
    private int userQueryLimit;
    
    @Value("${rate.limit.order-create:10}")
    private int orderCreateLimit;
    
    @Bean
    @RefreshScope
    public RateLimiter userQueryLimiter() {
        return RateLimiter.create(userQueryLimit);
    }
    
    @Bean
    @RefreshScope
    public RateLimiter orderCreateLimiter() {
        return RateLimiter.create(orderCreateLimit);
    }
}

┌───────────────┐
│ 关闭 (Closed) │◀────────────────────────────────┐
└───────────────┘                                 │
        │                                         │
        │ 失败率达到阈值                          │ 探测成功
        ▼                                         │
┌───────────────┐      超时窗口结束       ┌─────────────────────┐
│  打开 (Open)  │ ──────────────────────> │ 半开 (Half-Open)    │
└───────────────┘                         └─────────────────────┘
        ▲                                         │
        │              探测失败，重新打开          │
        └─────────────────────────────────────────┘

状态说明:
- Closed (关闭): 正常状态,请求正常通过
- Open (打开): 熔断状态,快速失败
- Half-Open (半开): 探测状态,少量请求通过探测下游是否恢复

# application.yml
resilience4j:
  circuitbreaker:
    configs:
      default:
        sliding-window-size: 100
        failure-rate-threshold: 50
        wait-duration-in-open-state: 10s
        permitted-number-of-calls-in-half-open-state: 10
        slow-call-duration-threshold: 2s
        slow-call-rate-threshold: 100
        minimum-number-of-calls: 10
    instances:
      userService:
        base-config: default
      orderService:
        sliding-window-size: 50
        failure-rate-threshold: 30

@Service
public class OrderService {
    
    @Autowired
    private UserClient userClient;
    
    // 方式1: 注解方式
    @CircuitBreaker(name = "userService", fallbackMethod = "getUserFallback")
    public User getUser(Long userId) {
        return userClient.getUser(userId);
    }
    
    // 降级方法
    public User getUserFallback(Long userId, Exception e) {
        log.warn("获取用户信息失败,使用降级数据: userId={}, error={}", userId, e.getMessage());
        return User.defaultUser(userId);
    }
    
    // 方式2: 编程方式
    public User getUserByCode(Long userId) {
        CircuitBreaker circuitBreaker = CircuitBreaker.ofDefaults("userService");
        
        Supplier<User> supplier = CircuitBreaker.decorateSupplier(
            circuitBreaker,
            () -> userClient.getUser(userId)
        );
        
        return Try.ofSupplier(supplier)
            .recover(throwable -> User.defaultUser(userId))
            .get();
    }
}

// 获取熔断器指标
@Component
public class CircuitBreakerMetrics {
    
    @EventListener
    public void onCircuitBreakerEvent(CircuitBreakerEvent event) {
        CircuitBreaker circuitBreaker = event.getCircuitBreaker();
        
        log.info("熔断器状态变更: name={}, state={}, failureRate={}, metrics={}",
            circuitBreaker.getName(),
            circuitBreaker.getState(),
            circuitBreaker.getMetrics().getFailureRate(),
            circuitBreaker.getMetrics()
        );
    }
}

// 资源定义
@GetMapping("/users/{id}")
@SentinelResource(value = "getUser", blockHandler = "handleBlock")
public User getUser(@PathVariable Long id) {
    return userService.getUser(id);
}

// 限流降级处理
public User handleBlock(Long id, BlockException ex) {
    return User.defaultUser(id);
}

// 限流规则
FlowRule flowRule = new FlowRule();
flowRule.setResource("getUser");
flowRule.setGrade(RuleConstant.FLOW_GRADE_QPS);
flowRule.setCount(100);

// 熔断规则
DegradeRule degradeRule = new DegradeRule();
degradeRule.setResource("getUser");
degradeRule.setGrade(CircuitBreakerStrategy.ERROR_RATIO.getType());
degradeRule.setCount(0.5); // 错误率50%
degradeRule.setTimeWindow(10); // 熔断时长10秒
degradeRule.setMinRequestAmount(10); // 最小请求数
degradeRule.setStatIntervalMs(1000); // 统计时长

// 加载规则
List<FlowRule> flowRules = new ArrayList<>();
flowRules.add(flowRule);
FlowRuleManager.loadRules(flowRules);

List<DegradeRule> degradeRules = new ArrayList<>();
degradeRules.add(degradeRule);
DegradeRuleManager.loadRules(degradeRules);

# application.yml
spring:
  cloud:
    sentinel:
      transport:
        dashboard: localhost:8080
        port: 8719
      eager: true # 立即初始化

@SentinelResource(value = "getUser", fallback = "getUserFallback")
public User getUser(Long userId) {
    return userClient.getUser(userId);
}

public User getUserFallback(Long userId) {
    return User.defaultUser(userId); // 返回默认用户
}

@SentinelResource(value = "getProduct", fallback = "getProductFallback")
public Product getProduct(Long productId) {
    return productClient.getProduct(productId);
}

public Product getProductFallback(Long productId) {
    return cacheService.getProductFromCache(productId); // 从缓存获取
}

@SentinelResource(value = "getProductDetail", fallback = "getProductDetailFallback")
public ProductDetail getProductDetail(Long productId) {
    return productClient.getProductDetail(productId);
}

public ProductDetail getProductDetailFallback(Long productId) {
    ProductDetail detail = new ProductDetail();
    detail.setId(productId);
    detail.setName("商品信息加载中");
    detail.setPrice(BigDecimal.ZERO);
    return detail; // 返回简化版商品信息
}

@GetMapping("/index")
public IndexPage getIndexPage() {
    IndexPage page = new IndexPage();
    
    // 核心功能:必须获取
    page.setProducts(productService.getProducts());
    
    // 非核心功能:失败时可以关闭
    try {
        page.setRecommendations(recommendationService.getRecommendations());
    } catch (Exception e) {
        log.warn("推荐服务不可用,跳过推荐模块");
        page.setRecommendations(Collections.emptyList());
    }
    
    return page;
}

# 熔断阈值设置建议
resilience4j:
  circuitbreaker:
    configs:
      default:
        # 错误率阈值: 50%
        failure-rate-threshold: 50
        
        # 慢调用比例阈值: 100%
        slow-call-rate-threshold: 100
        
        # 慢调用时间阈值: 2秒
        slow-call-duration-threshold: 2s
        
        # 滑动窗口大小: 100次调用
        sliding-window-size: 100
        
        # 最小调用次数: 10次
        minimum-number-of-calls: 10
        
        # 熔断持续时间: 10秒
        wait-duration-in-open-state: 10s
        
        # 半开状态允许的调用次数: 10次
        permitted-number-of-calls-in-half-open-state: 10

// 核心链路:严格熔断,必须有降级
@CircuitBreaker(name = "orderService", fallbackMethod = "createOrderFallback")
public Order createOrder(OrderRequest request) {
    // 核心业务逻辑
}

// 非核心链路:可以熔断,降级可以为空
@CircuitBreaker(name = "recommendationService")
public List<Product> getRecommendations(Long userId) {
    // 非核心业务逻辑
}

选择降级策略的决策树:

问题: 服务不可用时如何降级?
    ↓
是否有缓存数据?
    ├─ 是 → 返回缓存数据
    └─ 否 ↓
是否可以返回默认值?
    ├─ 是 → 返回默认值
    └─ 否 ↓
是否可以简化数据?
    ├─ 是 → 返回简化数据
    └─ 否 ↓
是否可以关闭功能?
    ├─ 是 → 关闭功能
    └─ 否 ↓
返回错误,触发告警

┌─────────────────────────────────────────────────────┐
│                    客户端请求                        │
└─────────────────────────────────────────────────────┘
                          ↓
┌─────────────────────────────────────────────────────┐
│  限流 (第一道防线)                                   │
│  - 入口削峰                                          │
│  - 保护系统不超载                                    │
└─────────────────────────────────────────────────────┘
                          ↓
┌─────────────────────────────────────────────────────┐
│  熔断 (第二道防线)                                   │
│  - 依赖隔离                                          │
│  - 快速失败                                          │
└─────────────────────────────────────────────────────┘
                          ↓
┌─────────────────────────────────────────────────────┐
│  重试 (第三道防线)                                   │
│  - 只对可恢复错误生效                                │
│  - 有边界限制                                        │
└─────────────────────────────────────────────────────┘
                          ↓
┌─────────────────────────────────────────────────────┐
│                  目标服务                            │
└─────────────────────────────────────────────────────┘

超时时间设置公式:

超时时间 = 平均响应时间 × 系数(2-3) + 网络延迟

示例:
- 平均响应时间: 100ms
- 系数: 2
- 网络延迟: 50ms
- 超时时间 = 100 × 2 + 50 = 250ms

# application.yml
spring:
  cloud:
    gateway:
      httpclient:
        connect-timeout: 3000 # 连接超时: 3秒
        response-timeout: 5000 # 响应超时: 5秒
    
feign:
  client:
    config:
      default:
        connectTimeout: 3000
        readTimeout: 5000

// Spring Cloud Gateway 重试配置
@Bean
public RouteLocator customRouteLocator(RouteLocatorBuilder builder) {
    return builder.routes()
        .route("user-service", r -> r.path("/api/users/**")
            .filters(f -> f
                .retry(config -> config
                    .setRetries(3) // 重试3次
                    .setStatuses(HttpStatus.INTERNAL_SERVER_ERROR) // 对500错误重试
                    .setMethods(HttpMethod.GET) // 只对GET请求重试
                    .setBackoff(Duration.ofMillis(100), Duration.ofMillis(1000), 2, true)
                )
            )
            .uri("lb://user-service"))
        .build();
}

// Feign 重试配置
@Configuration
public class FeignConfig {
    
    @Bean
    public Retryer feignRetryer() {
        // 最大重试次数: 3
        // 初始间隔: 100ms
        // 最大间隔: 1s
        return new Retryer.Default(100, TimeUnit.SECONDS.toMillis(1), 3);
    }
}

// 重试必须有边界限制
@Service
public class OrderService {
    
    @Autowired
    private UserClient userClient;
    
    // √ 正确: 有重试次数限制
    @Retryable(
        value = {ConnectException.class, TimeoutException.class},
        maxAttempts = 3,
        backoff = @Backoff(delay = 100, multiplier = 2)
    )
    public User getUser(Long userId) {
        return userClient.getUser(userId);
    }
    
    // × 错误: 无限重试
    public User getUserWrong(Long userId) {
        while (true) {
            try {
                return userClient.getUser(userId);
            } catch (Exception e) {
                // 无限重试,可能导致死循环
            }
        }
    }
}

# 多层限流配置
spring:
  cloud:
    gateway:
      routes:
        - id: seckill-service
          uri: lb://seckill-service
          predicates:
            - Path=/api/seckill/**
          filters:
            # 第一层:全局限流
            - name: RequestRateLimiter
              args:
                redis-rate-limiter.replenishRate: 1000
                redis-rate-limiter.burstCapacity: 5000
                key-resolver: "#{@ipKeyResolver}"
            
            # 第二层:用户级限流
            - name: RequestRateLimiter
              args:
                redis-rate-limiter.replenishRate: 10
                redis-rate-limiter.burstCapacity: 20
                key-resolver: "#{@userKeyResolver}"
            
            # 第三层:熔断保护
            - name: CircuitBreaker
              args:
                name: seckillCircuitBreaker
                fallbackUri: forward:/fallback/seckill

@Service
public class OrderService {
    
    @Autowired
    private MemberClient memberClient;
    
    // 熔断+降级
    @CircuitBreaker(name = "memberService", fallbackMethod = "getMemberBenefitFallback")
    public MemberBenefit getMemberBenefit(Long userId) {
        return memberClient.getBenefit(userId);
    }
    
    // 降级策略:返回默认权益
    public MemberBenefit getMemberBenefitFallback(Long userId, Exception ex) {
        log.warn("会员服务不可用,使用默认权益: userId={}", userId);
        
        MemberBenefit benefit = new MemberBenefit();
        benefit.setUserId(userId);
        benefit.setLevel(1); // 默认等级
        benefit.setDiscount(BigDecimal.ONE); // 无折扣
        benefit.setPoints(0); // 无积分
        
        return benefit;
    }
}

// 多维度限流
@Configuration
public class RateLimitConfig {
    
    // IP维度限流
    @Bean
    public KeyResolver ipKeyResolver() {
        return exchange -> Mono.just(
            exchange.getRequest().getRemoteAddress().getAddress().getHostAddress()
        );
    }
    
    // 用户维度限流
    @Bean
    public KeyResolver userKeyResolver() {
        return exchange -> Mono.just(
            exchange.getRequest().getHeaders().getFirst("X-User-Id")
        );
    }
    
    // 接口维度限流
    @Bean
    public KeyResolver apiKeyResolver() {
        return exchange -> Mono.just(
            exchange.getRequest().getPath().value()
        );
    }
    
    // 组合维度限流
    @Bean
    public KeyResolver compositeKeyResolver() {
        return exchange -> Mono.just(
            exchange.getRequest().getRemoteAddress().getAddress().getHostAddress() + ":" +
            exchange.getRequest().getPath().value()
        );
    }
}

// IP 黑名单
@Component
public class IpBlacklistFilter implements GlobalFilter, Ordered {
    
    @Autowired
    private RedisTemplate<String, String> redisTemplate;
    
    @Override
    public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
        String ip = exchange.getRequest().getRemoteAddress().getAddress().getHostAddress();
        
        // 检查黑名单
        Boolean isBlacklisted = redisTemplate.opsForSet().isMember("ip:blacklist", ip);
        if (isBlacklisted != null && isBlacklisted) {
            exchange.getResponse().setStatusCode(HttpStatus.FORBIDDEN);
            return exchange.getResponse().setComplete();
        }
        
        return chain.filter(exchange);
    }
    
    @Override
    public int getOrder() {
        return -300;
    }
}

# 检查路由配置
curl http://gateway:8080/actuator/gateway/routes

// 检查灰度规则
@Component
public class GrayRouteFilter implements GlobalFilter {
    
    @Override
    public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
        String grayVersion = exchange.getRequest().getHeaders().getFirst("X-Gray-Version");
        
        log.info("灰度规则: path={}, grayVersion={}", 
            exchange.getRequest().getPath(), grayVersion);
        
        return chain.filter(exchange);
    }
}

// 检查鉴权耗时
@Component
public class AuthFilter implements GlobalFilter {
    
    @Override
    public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
        long startTime = System.currentTimeMillis();
        
        // 鉴权逻辑
        boolean authenticated = authenticate(exchange);
        
        long duration = System.currentTimeMillis() - startTime;
        if (duration > 100) {
            log.warn("鉴权耗时过长: {}ms, path={}", 
                duration, exchange.getRequest().getPath());
        }
        
        if (!authenticated) {
            exchange.getResponse().setStatusCode(HttpStatus.UNAUTHORIZED);
            return exchange.getResponse().setComplete();
        }
        
        return chain.filter(exchange);
    }
}

# 检查网关健康状态
curl http://gateway:8080/actuator/health

# 检查网关指标
curl http://gateway:8080/actuator/metrics

# 检查限流配置
curl http://gateway:8080/actuator/gateway/routes

// 查看限流日志
@Component
public class RateLimitFilter implements GlobalFilter {
    
    @Override
    public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
        String ip = exchange.getRequest().getRemoteAddress().getAddress().getHostAddress();
        String path = exchange.getRequest().getPath().value();
        
        log.info("限流检查: ip={}, path={}", ip, path);
        
        return chain.filter(exchange);
    }
}

// 动态调整限流阈值
@Configuration
@RefreshScope
public class RateLimitConfig {
    
    @Value("${rate.limit.threshold:100}")
    private int threshold;
    
    @Bean
    @RefreshScope
    public RateLimiter rateLimiter() {
        return RateLimiter.create(threshold);
    }
}

# 区分限流配置
rate:
  limit:
    core:
      threshold: 1000
    non-core:
      threshold: 100

# 支持动态配置
spring:
  cloud:
    nacos:
      config:
        server-addr: localhost:8848
        file-extension: yaml
        shared-configs:
          - data-id: rate-limit.yaml
            refresh: true

// 根据监控数据设置阈值
// 建议先观察一段时间的历史数据
CircuitBreakerConfig config = CircuitBreakerConfig.custom()
    .failureRateThreshold(50) // 根据实际错误率调整
    .slowCallRateThreshold(100)
    .slowCallDurationThreshold(Duration.ofSeconds(2))
    .waitDurationInOpenState(Duration.ofSeconds(10))
    .slidingWindowSize(100)
    .minimumNumberOfCalls(10)
    .build();

// 配置半开状态
CircuitBreakerConfig config = CircuitBreakerConfig.custom()
    .permittedNumberOfCallsInHalfOpenState(10) // 半开状态允许10次探测
    .build();

// 熔断和重试配合使用
@Service
public class OrderService {
    
    @CircuitBreaker(name = "userService", fallbackMethod = "getUserFallback")
    @Retry(name = "userService", fallbackMethod = "getUserFallback")
    public User getUser(Long userId) {
        return userClient.getUser(userId);
    }
    
    public User getUserFallback(Long userId, Exception e) {
        return User.defaultUser(userId);
    }
}

// 核心链路降级策略
@Service
public class OrderService {
    
    @CircuitBreaker(name = "paymentService", fallbackMethod = "processPaymentFallback")
    public PaymentResult processPayment(PaymentRequest request) {
        return paymentClient.process(request);
    }
    
    // 核心链路降级:记录日志,人工处理
    public PaymentResult processPaymentFallback(PaymentRequest request, Exception e) {
        log.error("支付服务不可用,需要人工处理: orderId={}", request.getOrderId());
        
        // 记录到数据库,后续人工处理
        pendingPaymentRepository.save(new PendingPayment(request));
        
        // 返回处理中状态
        PaymentResult result = new PaymentResult();
        result.setStatus("PENDING");
        result.setMessage("支付处理中,请稍后查询");
        return result;
    }
}

# 差异化限流配置
rate:
  limit:
    core:
      threshold: 1000
      strategy: strict
    non-core:
      threshold: 500
      strategy: moderate
    edge:
      threshold: 100
      strategy: loose

业务功能 → 是否可降级?
    ├─ 是 → 是否有缓存?
    │   ├─ 是 → 返回缓存
    │   └─ 否 → 是否有默认值?
    │       ├─ 是 → 返回默认值
    │       └─ 否 → 是否可简化?
    │           ├─ 是 → 返回简化数据
    │           └─ 否 → 是否可关闭?
    │               ├─ 是 → 关闭功能
    │               └─ 否 → 返回错误
    └─ 否 → 触发告警,人工介入

网关 ≠ 简单转发

网关 = 统一入口 + 横切能力集中治理

包括:
- 路由转发
- 认证授权
- 限流熔断
- 灰度发布
- 日志监控
- 协议转换
- 安全防护

限流维度:
├─ 全局限流 - 保护整体系统
├─ 服务级限流 - 保护单个服务
├─ 接口级限流 - 保护核心接口
├─ 用户级限流 - 防止单个用户滥用
├─ IP级限流 - 防止恶意攻击
└─ 租户级限流 - B端系统租户隔离

1. 熔断打开
    ↓
2. 等待超时 (如10秒)
    ↓
3. 进入半开状态
    ↓
4. 少量探测请求 (如10次)
    ↓
5. 判断探测结果:
    ├─ 成功率高 → 关闭熔断
    └─ 成功率低 → 重新打开熔断

重试规则:
- 重试次数: 最多3次
- 重试条件: 只对可恢复错误重试
- 重试间隔: 指数退避 (100ms, 200ms, 400ms)
- 重试限制: 必须有熔断保护

核心链路 (如订单、支付):
- 严格熔断保护
- 必须有兜底方案
- 降级后仍需保证核心功能可用
- 触发告警,人工介入

非核心链路 (如推荐、评论):
- 可以完全关闭
- 降级后返回空数据或默认值
- 不影响核心业务流程

限流 (第一道防线):
- 入口削峰
- 保护系统不超载

熔断 (第二道防线):
- 依赖隔离
- 快速失败

重试 (第三道防线):
- 只对可恢复错误生效
- 有边界限制

三者配合:
- 限流防止系统过载
- 熔断防止依赖拖垮整体
- 重试只在合适场景使用

第一层: 网关全局限流
- 防止流量直接冲垮系统
- 阈值: 系统容量的1.5倍

第二层: 用户级限流
- 防止单个用户刷单
- 阈值: 每用户1次/秒

第三层: 商品级限流
- 保护库存扣减接口
- 阈值: 库存数量

第四层: 熔断保护
- 下游服务异常时快速失败
- 返回友好提示