分布式组件Gateway技术栈系统性理解和操作

// 1. 引入依赖（pom.xml）
<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-starter-gateway</artifactId>
</dependency>
// 2. 配置文件（application.yml）
spring:
  cloud:
    gateway:
      routes:
        - id: user-service
          uri: lb://user-service
          predicates:
            - Path=/api/user/**
          filters:
            - StripPrefix=1
// 3. 启动类
@SpringBootApplication
public class GatewayApplication {
    public static void main(String[] args) {
        SpringApplication.run(GatewayApplication.class, args);
    }
}

// 集成Nacos注册中心
<dependency>
    <groupId>com.alibaba.cloud</groupId>
    <artifactId>spring-cloud-starter-alibaba-nacos-discovery</artifactId>
</dependency>
// 配置Nacos
spring:
  cloud:
    nacos:
      discovery:
        server-addr: 127.0.0.1:8848
    gateway:
      discovery:
        locator:
          enabled: true  # 开启服务发现
          lower-case-service-id: true

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

// 多条件组合断言
spring:
  cloud:
    gateway:
      routes:
        - id: complex-route
          uri: lb://service
          predicates:
            - Path=/api/**
            - Method=GET
            - Header=X-Request-Id, \d+
            - Query=token, \w+
            - After=2024-01-01T00:00:00+08:00

// 自定义全局过滤器
@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();
        }
        return chain.filter(exchange);
    }
    @Override
    public int getOrder() {
        return -100; // 优先级
    }
}

// 自动集成Ribbon/LoadBalancer
spring:
  cloud:
    gateway:
      routes:
        - id: service-route
          uri: lb://service-name  # lb://前缀触发负载均衡

// 基于Redis的分布式限流
spring:
  cloud:
    gateway:
      routes:
        - id: rate-limit-route
          uri: lb://service
          filters:
            - name: RequestRateLimiter
              args:
                redis-rate-limiter.replenishRate: 10  # 令牌桶填充速率
                redis-rate-limiter.burstCapacity: 20  # 令牌桶容量
                key-resolver: "#{@userKeyResolver}"   # 自定义key解析器

@Bean
KeyResolver userKeyResolver() {
    return exchange -> Mono.just(exchange.getRequest().getQueryParams().getFirst("userId"));
}

// 集成Hystrix/Sentinel
spring:
  cloud:
    gateway:
      routes:
        - id: circuit-breaker-route
          uri: lb://service
          filters:
            - name: Hystrix
              args:
                name: fallbackCommand
                fallbackUri: forward:/fallback

// 全局CORS配置
@Configuration
public class CorsConfig {
    @Bean
    public CorsWebFilter corsWebFilter() {
        CorsConfiguration config = new CorsConfiguration();
        config.addAllowedOrigin("*");
        config.addAllowedMethod("*");
        config.addAllowedHeader("*");
        config.setAllowCredentials(true);
        UrlBasedCorsConfigurationSource source = new UrlBasedCorsConfigurationSource();
        source.registerCorsConfiguration("/**", config);
        return new CorsWebFilter(source);
    }
}

// 综合应用：API网关的完整配置
spring:
  cloud:
    gateway:
      # 全局过滤器配置
      default-filters:
        - AddResponseHeader=X-Response-Default, Default-Value
      routes:
        # 用户服务路由
        - id: user-service
          uri: lb://user-service
          predicates:
            - Path=/api/user/**
          filters:
            - StripPrefix=1
            - name: RequestRateLimiter
              args:
                redis-rate-limiter.replenishRate: 100
                redis-rate-limiter.burstCapacity: 200
        # 订单服务路由（带认证）
        - id: order-service
          uri: lb://order-service
          predicates:
            - Path=/api/order/**
          filters:
            - StripPrefix=1
            - name: Hystrix
              args:
                name: orderFallback
                fallbackUri: forward:/order/fallback
        # 静态资源路由
        - id: static-resources
          uri: http://static.example.com
          predicates:
            - Path=/static/**

// 1. DispatcherHandler - 请求入口分发器
public class DispatcherHandler implements WebHandler {
    private final Map<String, HandlerMapping> handlerMappings;
    @Override
    public Mono<Void> handle(ServerWebExchange exchange) {
        // 遍历所有HandlerMapping，找到匹配的处理器
        return handlerMappings.values()
            .stream()
            .filter(mapping -> mapping.matches(exchange))
            .findFirst()
            .map(mapping -> mapping.getHandler(exchange))
            .orElse(Mono.error(new ResponseStatusException(HttpStatus.NOT_FOUND)))
            .flatMap(handler -> handler.handle(exchange));
    }
}
// 2. RoutePredicateHandlerMapping - 路由匹配器
public class RoutePredicateHandlerMapping extends AbstractHandlerMapping {
    @Override
    protected Mono<?> getHandlerInternal(ServerWebExchange exchange) {
        // 1. 获取所有路由
        Flux<Route> routes = routeLocator.getRoutes();
        // 2. 根据Predicate匹配路由
        return routes.filter(route -> {
            List<PredicateDefinition> predicates = route.getPredicates();
            return predicates.stream()
                .allMatch(predicate -> predicate.apply(exchange));
        })
        .next() // 返回第一个匹配的路由
        .map(route -> {
            exchange.getAttributes().put(GATEWAY_ROUTE_ATTR, route);
            return webHandler; // 返回FilteringWebHandler
        });
    }
}
// 3. FilteringWebHandler - 过滤器链处理器
public class FilteringWebHandler implements WebHandler {
    @Override
    public Mono<Void> handle(ServerWebExchange exchange) {
        Route route = exchange.getAttribute(GATEWAY_ROUTE_ATTR);
        // 1. 获取路由的过滤器
        List<GatewayFilter> routeFilters = route.getFilters();
        // 2. 获取全局过滤器
        List<GlobalFilter> globalFilters = globalFilters();
        // 3. 合并并排序
        List<GatewayFilter> combined = combineFilters(routeFilters, globalFilters);
        // 4. 构建过滤器链
        GatewayFilterChain chain = new DefaultGatewayFilterChain(combined);
        // 5. 执行过滤器链
        return chain.filter(exchange);
    }
}
// 4. DefaultGatewayFilterChain - 责任链模式
private static class DefaultGatewayFilterChain implements GatewayFilterChain {
    private final int index;
    private final List<GatewayFilter> filters;
    @Override
    public Mono<Void> filter(ServerWebExchange exchange) {
        if (index < filters.size()) {
            // 递归调用下一个过滤器
            GatewayFilter filter = filters.get(index);
            return filter.filter(exchange, this);
        } else {
            // 所有过滤器执行完毕，转发请求
            return ((NettyRoutingFilter) filters.get(filters.size() - 1))
                .route(exchange);
        }
    }
}

// 传统阻塞IO vs 响应式非阻塞IO
// 传统Servlet模型（阻塞）
@WebServlet("/api")
public class BlockingServlet extends HttpServlet {
    @Override
    protected void doGet(HttpServletRequest req, HttpServletResponse resp) {
        // 每个请求占用一个线程，直到IO完成
        String result = database.query(); // 阻塞等待
        resp.getWriter().write(result);
    }
}
// 问题：1000个并发请求 = 1000个线程 = 高内存消耗
// Gateway响应式模型（非阻塞）
@GetMapping("/api")
public Mono<String> reactiveApi() {
    // 线程不等待，注册回调，IO完成后触发
    return database.queryAsync() // 返回Mono
        .map(result -> process(result));
}
// 优势：少量线程处理大量并发请求

// Netty的内存池化技术（PooledByteBufAllocator）
// 传统方式：每次分配新内存
byte[] buffer1 = new byte[1024]; // 分配
// 使用...
buffer1 = null; // 释放，等待GC
// Netty内存池：复用内存块
PooledByteBufAllocator allocator = PooledByteBufAllocator.DEFAULT;
ByteBuf buffer = allocator.buffer(1024); // 从池中获取
// 使用...
buffer.release(); // 归还到池中，可复用
// 优势：
// 1. 减少内存分配/释放开销
// 2. 降低GC压力
// 3. 避免内存碎片

// HTTP连接池配置
spring:
  cloud:
    gateway:
      httpclient:
        pool:
          type: ELASTIC  # 弹性连接池
          max-connections: 1000  # 最大连接数
          acquire-timeout: 45000 # 获取连接超时
// 连接复用机制
┌─────────────────────────────────────────────────────┐
│  Connection Pool (连接池)                            │
│  ┌─────────┐  ┌─────────┐  ┌─────────┐             │
│  │ Conn1   │  │ Conn2   │  │ Conn3   │             │
│  │ (复用)   │  │ (复用)   │  │ (复用)   │             │
│  └─────────┘  └─────────┘  └─────────┘             │
└─────────────────────────────────────────────────────┘
        ↓           ↓           ↓
    Service1    Service2    Service3
// 优势：
// 1. 避免频繁建立/断开TCP连接
// 2. 减少TCP握手开销
// 3. 提高请求响应速度

// 路由匹配流程
public class RoutePredicateHandlerMapping {
    public Mono<Route> getRoute(ServerWebExchange exchange) {
        return routeLocator.getRoutes()
            .filter(route -> {
                // 1. Path匹配（AntPathMatcher）
                if (route.getPath() != null) {
                    AntPathMatcher matcher = new AntPathMatcher();
                    if (!matcher.match(route.getPath(), exchange.getRequest().getPath().value())) {
                        return false;
                    }
                }
                // 2. Method匹配
                if (route.getMethod() != null) {
                    if (!route.getMethod().equals(exchange.getRequest().getMethod())) {
                        return false;
                    }
                }
                // 3. Header匹配
                if (route.getHeaders() != null) {
                    for (Map.Entry<String, String> header : route.getHeaders().entrySet()) {
                        String value = exchange.getRequest().getHeaders().getFirst(header.getKey());
                        if (value == null || !value.matches(header.getValue())) {
                            return false;
                        }
                    }
                }
                // 4. Query匹配
                // 5. Cookie匹配
                // 6. 其他Predicate...
                return true;
            })
            .next(); // 返回第一个匹配的路由
    }
}
// 性能优化：路由缓存
private final Map<String, Route> routeCache = new ConcurrentHashMap<>();
public Mono<Route> getCachedRoute(String path) {
    return Mono.justOrEmpty(routeCache.get(path))
        .switchIfEmpty(getRouteFromLocator(path)
            .doOnNext(route -> routeCache.put(path, route)));
}

// LoadBalancerClientFilter - 负载均衡过滤器
public class LoadBalancerClientFilter implements GlobalFilter {
    @Override
    public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
        URI url = exchange.getAttribute(GATEWAY_REQUEST_URL_ATTR);
        if (url != null && "lb".equals(url.getScheme())) {
            // 1. 解析服务名
            String serviceId = url.getHost();
            // 2. 获取服务实例列表
            ServiceInstance instance = loadBalancer.choose(serviceId);
            // 3. 构建实际请求URL
            URI uri = reconstructURI(instance, url);
            exchange.getAttributes().put(GATEWAY_REQUEST_URL_ATTR, uri);
        }
        return chain.filter(exchange);
    }
}
// 负载均衡算法实现
public class RoundRobinLoadBalancer implements ReactorServiceInstanceLoadBalancer {
    private final AtomicInteger position = new AtomicInteger();
    @Override
    public Mono<ServiceInstance> choose(Request request) {
        List<ServiceInstance> instances = discoveryClient.getInstances(serviceId);
        if (instances.isEmpty()) {
            return Mono.empty();
        }
        // 轮询算法
        int pos = Math.abs(position.incrementAndGet() % instances.size());
        return Mono.just(instances.get(pos));
    }
}
// 权重负载均衡
public class WeightedLoadBalancer {
    public ServiceInstance chooseWeighted(List<ServiceInstance> instances) {
        // 计算总权重
        int totalWeight = instances.stream()
            .mapToInt(i -> i.getMetadata().getOrDefault("weight", "1"))
            .mapToInt(Integer::parseInt)
            .sum();
        // 随机选择
        int random = ThreadLocalRandom.current().nextInt(totalWeight);
        // 根据权重分配
        int current = 0;
        for (ServiceInstance instance : instances) {
            current += Integer.parseInt(instance.getMetadata().get("weight"));
            if (random < current) {
                return instance;
            }
        }
        return instances.get(0);
    }
}

// 基于Nacos的动态路由
@Component
public class NacosRouteDefinitionRepository implements RouteDefinitionRepository {
    @Autowired
    private ConfigService configService;
    // 监听Nacos配置变化
    @PostConstruct
    public void init() throws NacosException {
        configService.addListener("gateway-routes", "DEFAULT_GROUP", 
            new Listener() {
                @Override
                public void receiveConfigInfo(String configInfo) {
                    // 解析配置
                    List<RouteDefinition> routes = parseRoutes(configInfo);
                    // 更新路由
                    routes.forEach(route -> {
                        routeDefinitionWriter.save(Mono.just(route)).subscribe();
                    });
                }
            });
    }
    @Override
    public Flux<RouteDefinition> getRouteDefinitions() {
        // 从Nacos获取路由配置
        String config = configService.getConfig("gateway-routes", "DEFAULT_GROUP", 5000);
        List<RouteDefinition> routes = parseRoutes(config);
        return Flux.fromIterable(routes);
    }
}
// 路由热更新流程
┌─────────────────────────────────────────────────────┐
│  1. Nacos配置变更                                     │
│     ↓                                                │
│  2. 监听器收到通知                                    │
│     ↓                                                │
│  3. 解析新的路由配置                                  │
│     ↓                                                │
│  4. 调用RouteDefinitionWriter保存                     │
│     ↓                                                │
│  5. CachingRouteLocator清除缓存                       │
│     ↓                                                │
│  6. 新路由生效（无需重启）                            │
└─────────────────────────────────────────────────────┘