Spring Cloud Gateway WebFlux 模式架构分析

技术分享 2小时前 0 999+

Spring Cloud Gateway WebFlux 模式架构分析

请关注微信公众号：阿呆-bot

1. 入口类及关键类关系

1.1 入口类

Spring Cloud Gateway WebFlux 模式的入口类是 GatewayAutoConfiguration，这是整个 WebFlux 模式的"总指挥"。它负责配置所有 Gateway 需要的组件。

@Configuration(proxyBeanMethods = false) @ConditionalOnProperty(name = "spring.cloud.gateway.server.webflux.enabled", matchIfMissing = true) @EnableConfigurationProperties @AutoConfigureBefore({ HttpHandlerAutoConfiguration.class, WebFluxAutoConfiguration.class }) @AutoConfigureAfter({ GatewayReactiveLoadBalancerClientAutoConfiguration.class,         GatewayClassPathWarningAutoConfiguration.class }) @ConditionalOnClass(DispatcherHandler.class) public class GatewayAutoConfiguration {     // 核心 Bean 定义 }

关键特性说明：

@AutoConfigureBefore({ HttpHandlerAutoConfiguration.class, WebFluxAutoConfiguration.class })
- 这个很关键！Gateway 必须在 WebFlux 的默认配置之前加载
- 为什么？因为 Gateway 需要注册自己的 HandlerMapping（RoutePredicateHandlerMapping）
- 如果 WebFlux 的默认配置先加载，可能会注册其他的 HandlerMapping，导致路由冲突
- Gateway 的 HandlerMapping 优先级更高，会优先处理请求
@ConditionalOnClass(DispatcherHandler.class)
- 这个注解确保只有在 WebFlux 存在时才加载 Gateway
- DispatcherHandler 是 WebFlux 的核心组件，类似于 WebMVC 的 DispatcherServlet
- 如果没有 WebFlux，Gateway WebFlux 模式就无法工作

@ConditionalOnProperty

# 可以通过配置控制是否启用 spring:   cloud:     gateway:       server:         webflux:           enabled: true  # 默认启用，可以设置为 false 禁用

配置的核心组件：

RouteLocator：路由定位器，负责提供路由列表
FilteringWebHandler：过滤器处理器，负责执行过滤器链
RoutePredicateHandlerMapping：路由匹配器，负责匹配请求到路由
GatewayProperties：Gateway 配置属性

1.2 关键类关系图

1.3 核心组件说明

RoutePredicateHandlerMapping

这是 Gateway 的路由匹配器，负责找到匹配的路由。它继承自 Spring WebFlux 的 AbstractHandlerMapping。

工作原理：

@Override protected Mono<?> getHandlerInternal(ServerWebExchange exchange) {     // 1. 查找匹配的路由（异步操作）     return lookupRoute(exchange)         .map(route -> {             // 2. 把路由信息存储到 Exchange 的属性中             exchange.getAttributes().put(GATEWAY_ROUTE_ATTR, route);             // 3. 返回 FilteringWebHandler 作为处理器             return webHandler;         }); }

实际例子：

// 假设有这些路由配置 routes = [     { id: "user-service", path: "/api/users/**", uri: "http://user-service" },     { id: "order-service", path: "/api/orders/**", uri: "http://order-service" } ]  // 请求：GET /api/users/123 // lookupRoute() 会： // 1. 遍历所有路由 // 2. 使用 AsyncPredicate 异步匹配 // 3. 找到匹配的路由：user-service // 4. 返回 FilteringWebHandler

关键点：

lookupRoute() 返回 Mono<Route>，是异步的
路由匹配使用 AsyncPredicate，支持异步条件判断
匹配成功后，路由信息存储在 ServerWebExchange 的属性中

FilteringWebHandler

这是 Gateway 的核心请求处理器，负责执行过滤器链。它实现了 Spring WebFlux 的 WebHandler 接口。

工作原理：

@Override public Mono<Void> handle(ServerWebExchange exchange) {     // 1. 从 Exchange 中获取路由信息（路由匹配时设置的）     Route route = exchange.getRequiredAttribute(GATEWAY_ROUTE_ATTR);          // 2. 获取合并后的过滤器列表（全局过滤器 + 路由过滤器）     List<GatewayFilter> combined = getCombinedFilters(route);          // 3. 创建过滤器链并执行     return new DefaultGatewayFilterChain(combined).filter(exchange); }

实际例子：

// 假设路由配置了这些过滤器 route.filters = [     AddRequestHeaderFilter,     RateLimitFilter,     NettyRoutingFilter  // 最后一个，负责实际转发请求 ]  // FilteringWebHandler 会： // 1. 合并全局过滤器和路由过滤器 // 2. 按 Order 排序 // 3. 创建过滤器链 // 4. 依次执行过滤器

过滤器链执行：

// 过滤器链的实现（责任链模式） private static class DefaultGatewayFilterChain implements GatewayFilterChain {     @Override     public Mono<Void> filter(ServerWebExchange exchange) {         return Mono.defer(() -> {             if (this.index < filters.size()) {                 GatewayFilter filter = filters.get(this.index);                 // 创建下一个链节点                 DefaultGatewayFilterChain chain = new DefaultGatewayFilterChain(this, this.index + 1);                 // 执行当前过滤器，传入下一个链节点                 return filter.filter(exchange, chain);             }             return Mono.empty();  // 所有过滤器执行完毕         });     } }

RouteLocator

路由定位器，负责提供路由列表。Gateway 支持多个 RouteLocator，可以组合使用。

工作原理：

@Bean @Primary public RouteLocator cachedCompositeRouteLocator(List<RouteLocator> routeLocators) {     // 1. 组合多个 RouteLocator     CompositeRouteLocator composite = new CompositeRouteLocator(         Flux.fromIterable(routeLocators)     );          // 2. 添加缓存（提高性能）     return new CachingRouteLocator(composite); }

实际例子：

// 可以有多个 RouteLocator RouteLocator[] locators = [     PropertiesRouteDefinitionLocator,      // 从配置文件读取路由     DiscoveryClientRouteDefinitionLocator, // 从服务发现读取路由     InMemoryRouteDefinitionRepository      // 从内存（API）读取路由 ]  // CompositeRouteLocator 会合并所有路由 // 返回 Flux<Route>，包含所有路由

路由来源：

配置文件：PropertiesRouteDefinitionLocator

spring:   cloud:     gateway:       routes:         - id: user-service           uri: http://localhost:8081

服务发现：DiscoveryClientRouteDefinitionLocator

// 自动从 Eureka/Consul 等服务注册中心发现服务 // 为每个服务自动创建路由

API 动态配置：InMemoryRouteDefinitionRepository

// 通过 API 动态添加/删除路由 routeDefinitionRepository.save(routeDefinition);

2. 关键流程描述

2.1 请求处理流程时序图

2.2 路由匹配流程

路由匹配是 Gateway 的核心功能。让我们用一个实际例子来说明整个流程：

场景：客户端请求 GET /api/users/123

详细流程：

DispatcherHandler 接收 HTTP 请求
```
客户端 → DispatcherHandler → Gateway 
```
- DispatcherHandler 是 WebFlux 的请求分发器，类似于 WebMVC 的 DispatcherServlet
- 它会把请求交给合适的 HandlerMapping 处理
- Gateway 的 RoutePredicateHandlerMapping 优先级很高，会优先处理

RoutePredicateHandlerMapping 从 RouteLocator 获取所有路由

// 获取所有路由（返回 Flux<Route>，是响应式的） Flux<Route> routes = routeLocator.getRoutes();  // 实际的路由列表可能是： // [ //   { id: "user-service", path: "/api/users/**", uri: "http://user-service" }, //   { id: "order-service", path: "/api/orders/**", uri: "http://order-service" } // ]

RouteLocator 可能从多个来源获取路由（配置文件、服务发现、API）
返回的是 Flux<Route>，是响应式的流

使用 AsyncPredicate 异步匹配路由

// 异步匹配路由 return routes     .filterWhen(route -> {         // 使用 AsyncPredicate 异步匹配         return route.getPredicate().test(exchange);     })     .next();  // 返回第一个匹配的路由

匹配过程：

// 路由1：user-service AsyncPredicate predicate1 = exchange -> {     String path = exchange.getRequest().getPath().value();     return Mono.just(path.startsWith("/api/users/"));  // 异步返回 true/false }; // 路径 "/api/users/123" 匹配 "/api/users/**" ✓  // 路由2：order-service AsyncPredicate predicate2 = exchange -> {     String path = exchange.getRequest().getPath().value();     return Mono.just(path.startsWith("/api/orders/"));  // 异步返回 false }; // 路径 "/api/users/123" 不匹配 "/api/orders/**" ✗

使用 AsyncPredicate 异步匹配，支持异步条件判断
可以在这里做异步操作（比如查询 Redis、数据库等）
返回 Mono<Boolean>，不会阻塞线程

找到匹配的路由后，将路由信息存储到 ServerWebExchange 的属性中

// 匹配成功后 Route matchedRoute = ...;  // user-service 路由  // 存储路由信息到 Exchange 的属性中 exchange.getAttributes().put(GATEWAY_ROUTE_ATTR, matchedRoute); exchange.getAttributes().put(GATEWAY_REQUEST_URL_ATTR, matchedRoute.getUri());

路由信息存储在 ServerWebExchange 的属性中
后续的过滤器可以从这里获取路由信息
比如 NettyRoutingFilter 会从这里获取目标 URI

返回 FilteringWebHandler 作为处理器
```
// 返回 FilteringWebHandler return Mono.just(webHandler); 
```
- FilteringWebHandler 会负责执行过滤器链
- 最终会调用 NettyRoutingFilter 转发请求到下游服务

完整示例：

# 配置文件 spring:   cloud:     gateway:       routes:         - id: user-service           uri: http://user-service:8081           predicates:             - Path=/api/users/**             - Method=GET

匹配过程：

请求 GET /api/users/123 到达
路径 /api/users/123 匹配 /api/users/** ✓
方法 GET 匹配 GET ✓
找到匹配的路由 user-service
返回 FilteringWebHandler，目标 URI 是 http://user-service:8081

2.3 过滤器链执行流程

过滤器链是 Gateway 的核心机制。让我们用一个实际例子来说明：

场景：请求已经匹配到路由，现在需要执行过滤器链

详细流程：

FilteringWebHandler 从路由中获取过滤器列表

Route route = exchange.getRequiredAttribute(GATEWAY_ROUTE_ATTR); List<GatewayFilter> routeFilters = route.getFilters(); // 假设路由配置了这些过滤器： // [AddRequestHeaderFilter, RateLimitFilter]

从路由中获取路由级别的过滤器
这些过滤器只应用于当前路由

合并全局过滤器和路由过滤器，按 Order 排序

// 全局过滤器（应用于所有路由） List<GlobalFilter> globalFilters = [     LoadBalancerClientFilter,      // Order: 10100     NettyRoutingFilter,            // Order: Ordered.LOWEST_PRECEDENCE     NettyWriteResponseFilter       // Order: Ordered.LOWEST_PRECEDENCE - 1 ];  // 路由过滤器 List<GatewayFilter> routeFilters = [     AddRequestHeaderFilter,         // Order: 1000     RateLimitFilter                 // Order: -100 ];  // 合并并排序 List<GatewayFilter> combined = [     RateLimitFilter,                // -100 (最高优先级)     AddRequestHeaderFilter,         // 1000     LoadBalancerClientFilter,       // 10100     NettyRoutingFilter,             // LOWEST_PRECEDENCE     NettyWriteResponseFilter        // LOWEST_PRECEDENCE - 1 ];

全局过滤器 + 路由过滤器 = 合并后的过滤器列表
按 Order 排序，数字越小优先级越高
Ordered.LOWEST_PRECEDENCE 是最大整数，优先级最低

创建 DefaultGatewayFilterChain，实现责任链模式

// 创建过滤器链 DefaultGatewayFilterChain chain = new DefaultGatewayFilterChain(combined);  // 过滤器链的内部实现 private static class DefaultGatewayFilterChain implements GatewayFilterChain {     private final int index;  // 当前执行的过滤器索引     private final List<GatewayFilter> filters;      @Override     public Mono<Void> filter(ServerWebExchange exchange) {         return Mono.defer(() -> {             if (this.index < filters.size()) {                 GatewayFilter filter = filters.get(this.index);                 // 创建下一个链节点                 DefaultGatewayFilterChain nextChain =                      new DefaultGatewayFilterChain(this, this.index + 1);                 // 执行当前过滤器，传入下一个链节点                 return filter.filter(exchange, nextChain);             }             return Mono.empty();  // 所有过滤器执行完毕         });     } }

使用责任链模式，每个过滤器都可以决定是否继续执行下一个过滤器
通过递归创建链节点来实现

递归执行过滤器

// 执行流程（简化版） RateLimitFilter.filter(exchange, chain1)   -> 检查限流   -> chain1.filter(exchange)  // 调用下一个过滤器     -> AddRequestHeaderFilter.filter(exchange, chain2)       -> 添加请求头       -> chain2.filter(exchange)  // 调用下一个过滤器         -> LoadBalancerClientFilter.filter(exchange, chain3)           -> 负载均衡选择实例           -> chain3.filter(exchange)             -> NettyRoutingFilter.filter(exchange, chain4)               -> 转发请求到下游服务               -> chain4.filter(exchange)                 -> NettyWriteResponseFilter.filter(exchange, chain5)                   -> 写入响应                   -> chain5.filter(exchange)                     -> Mono.empty()  // 所有过滤器执行完毕

每个过滤器调用 chain.filter(exchange) 来执行下一个过滤器
如果某个过滤器不想继续执行，可以不调用 chain.filter()
比如限流过滤器，如果限流了，就直接返回错误，不继续执行

NettyRoutingFilter 作为最后一个过滤器，执行实际的 HTTP 请求

// NettyRoutingFilter 的实现（简化版） @Override public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {     URI requestUrl = exchange.getRequiredAttribute(GATEWAY_REQUEST_URL_ATTR);      // 使用 Reactor Netty HttpClient 发送请求（非阻塞）     return httpClient.get()         .uri(requestUrl)         .send((req, nettyOutbound) -> nettyOutbound.send(request.getBody()))         .responseConnection((res, connection) -> {             // 设置响应到 Exchange             exchange.getAttributes().put(CLIENT_RESPONSE_ATTR, res);             return Mono.just(res);         })         .then(chain.filter(exchange));  // 继续执行下一个过滤器 }

NettyRoutingFilter 使用 Reactor Netty 的 HttpClient 发送请求
这是非阻塞的，线程不会在这里等待
响应到达后，会继续执行下一个过滤器（NettyWriteResponseFilter）

响应通过 Mono 链式返回

// 响应返回流程 NettyWriteResponseFilter.filter(exchange, chain)   -> 写入响应到客户端   -> Mono.empty()  // 完成   -> 返回给 NettyRoutingFilter     -> 返回给 LoadBalancerClientFilter       -> 返回给 AddRequestHeaderFilter         -> 返回给 RateLimitFilter           -> 返回给 FilteringWebHandler             -> 返回给 RoutePredicateHandlerMapping               -> 返回给 DispatcherHandler                 -> 返回给客户端

响应通过 Mono 链式返回
每个过滤器都可以修改响应
最终返回给客户端

关键点：

过滤器链是响应式的，不会阻塞线程
每个过滤器都可以修改请求或响应
过滤器可以决定是否继续执行下一个过滤器
NettyRoutingFilter 是最后一个过滤器，负责实际转发请求

2.4 响应式 HTTP 请求

Flux<HttpClientResponse> responseFlux = httpClient     .headers(headers -> headers.add(httpHeaders))     .request(method)     .uri(url)     .send((req, nettyOutbound) -> nettyOutbound.send(request.getBody()))     .responseConnection((res, connection) -> {         // 处理响应         return Mono.just(res);     });

关键点：

使用 Reactor Netty 的非阻塞 HTTP 客户端
基于 Reactive Streams 的响应式编程模型
支持背压（Backpressure）控制

3. 实现关键点说明

3.1 响应式编程模型

WebFlux 模式完全基于响应式编程，使用 Project Reactor 的 Mono 和 Flux。这是 WebFlux 和 WebMVC 最根本的区别。

基本概念：

public interface GlobalFilter {     Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain); }

Mono 和 Flux 是什么？

Mono：表示 0 或 1 个元素的异步序列

Mono<String> name = Mono.just("Spring");  // 一个值 Mono<String> empty = Mono.empty();        // 0 个值 Mono<String> async = Mono.fromCallable(() -> {     // 异步操作     return fetchNameFromDatabase(); });

Flux：表示 0 到 N 个元素的异步序列

Flux<String> names = Flux.just("Spring", "Cloud", "Gateway");  // 多个值 Flux<String> stream = Flux.interval(Duration.ofSeconds(1))     .map(i -> "Event " + i);  // 无限流

实际例子：

// WebMVC 方式（阻塞） public ServerResponse handle(ServerRequest request) {     User user = userService.getUser(userId);  // 阻塞等待     return ServerResponse.ok().body(user); }  // WebFlux 方式（非阻塞） public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {     return userService.getUser(userId)  // 返回 Mono<User>，不阻塞         .flatMap(user -> {             // 用户数据到达后才执行这里             exchange.getAttributes().put("user", user);             return chain.filter(exchange);         }); }

优势说明：

非阻塞 I/O，提高并发性能

// 非阻塞调用下游服务 return httpClient.get()     .uri("http://backend-service/api/data")     .retrieve()     .bodyToMono(String.class)  // 返回 Mono，不阻塞线程     .flatMap(data -> {         // 数据到达后才执行这里         return processData(data);     }); // 在等待响应的这段时间，线程可以处理其他请求

线程不会阻塞，可以处理其他请求
一个线程可以处理多个请求
并发性能大幅提升

支持背压，防止内存溢出

// 背压示例 Flux<String> dataStream = getDataStream();  // 数据流  dataStream     .limitRate(100)  // 限制速率，防止内存溢出     .subscribe(data -> {         // 处理数据         processData(data);     });

如果生产者生产数据太快，消费者可以告诉生产者慢一点
防止内存溢出
这是 Reactive Streams 规范的核心特性

资源利用率高，少量线程处理大量请求

场景：处理 10000 并发请求  WebMVC: 需要 500 线程，内存 ~500MB WebFlux: 只需要 16 线程，内存 ~100MB  资源利用率提升 5 倍！

3.2 过滤器链模式

Gateway 使用责任链模式实现过滤器链。这是 Gateway 的核心设计模式之一。

责任链模式：

private static class DefaultGatewayFilterChain implements GatewayFilterChain {     private final int index;  // 当前过滤器索引     private final List<GatewayFilter> filters;          @Override     public Mono<Void> filter(ServerWebExchange exchange) {         return Mono.defer(() -> {             if (this.index < filters.size()) {                 GatewayFilter filter = filters.get(this.index);                 // 创建下一个链节点                 DefaultGatewayFilterChain chain =                      new DefaultGatewayFilterChain(this, this.index + 1);                 // 执行当前过滤器，传入下一个链节点                 return filter.filter(exchange, chain);             }             return Mono.empty();  // 所有过滤器执行完毕         });     } }

实际例子：

// 假设有这些过滤器 List<GatewayFilter> filters = [     RateLimitFilter,        // index 0     AddRequestHeaderFilter,  // index 1     NettyRoutingFilter      // index 2 ];  // 执行流程 chain0.filter(exchange)  // index = 0   -> RateLimitFilter.filter(exchange, chain1)  // index = 1     -> 检查限流     -> chain1.filter(exchange)       -> AddRequestHeaderFilter.filter(exchange, chain2)  // index = 2         -> 添加请求头         -> chain2.filter(exchange)           -> NettyRoutingFilter.filter(exchange, chain3)  // index = 3             -> 转发请求             -> chain3.filter(exchange)               -> index (3) >= filters.size()，返回 Mono.empty()

过滤器类型：

GlobalFilter - 全局过滤器

@Component public class CustomGlobalFilter implements GlobalFilter, Ordered {     @Override     public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {         // 这个过滤器会应用于所有路由         return chain.filter(exchange);     }      @Override     public int getOrder() {         return -1;  // 高优先级     } }

应用于所有路由
比如认证、日志、监控等

GatewayFilter - 路由过滤器

// 通过配置定义的路由过滤器 spring:   cloud:     gateway:       routes:         - id: user-service           filters:             - AddRequestHeader=X-User-Id, 123  # 路由过滤器

只应用于特定路由
通过配置文件或 API 定义

Ordered - 控制执行顺序

// 通过 Order 接口控制执行顺序 public class RateLimitFilter implements GlobalFilter, Ordered {     @Override     public int getOrder() {         return -100;  // 数字越小，优先级越高     } }  public class LoggingFilter implements GlobalFilter, Ordered {     @Override     public int getOrder() {         return 1000;  // 优先级较低     } }  // 执行顺序：RateLimitFilter (-100) -> LoggingFilter (1000)

过滤器可以做什么？

修改请求

@Override public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {     ServerHttpRequest request = exchange.getRequest();     ServerHttpRequest modifiedRequest = request.mutate()         .header("X-Request-Id", UUID.randomUUID().toString())         .build();     return chain.filter(exchange.mutate().request(modifiedRequest).build()); }

修改响应

@Override public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {     return chain.filter(exchange).then(Mono.fromRunnable(() -> {         ServerHttpResponse response = exchange.getResponse();         response.getHeaders().add("X-Response-Time",              String.valueOf(System.currentTimeMillis()));     })); }

中断执行

@Override public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {     if (isRateLimited(exchange)) {         // 限流了，直接返回错误，不继续执行         exchange.getResponse().setStatusCode(HttpStatus.TOO_MANY_REQUESTS);         return exchange.getResponse().setComplete();     }     return chain.filter(exchange);  // 继续执行 }

3.3 异步谓词（AsyncPredicate）

路由匹配使用异步谓词，支持异步条件判断：

public interface AsyncPredicate<T> {     Mono<Boolean> test(T t); }

支持的谓词：

Path Route Predicate
Method Route Predicate
Header Route Predicate
Host Route Predicate
Query Route Predicate
RemoteAddr Route Predicate
Weight Route Predicate

3.4 Netty 集成

使用 Reactor Netty 作为底层 HTTP 客户端，提供高性能的非阻塞 I/O：

public class NettyRoutingFilter implements GlobalFilter {     private final HttpClient httpClient;          @Override     public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {         // 使用 Reactor Netty HttpClient 发送请求         Flux<HttpClientResponse> responseFlux = httpClient             .request(method)             .uri(url)             .send(...)             .responseConnection(...);     } }

特性：

基于 Netty 的事件驱动模型
支持 HTTP/1.1 和 HTTP/2
支持 WebSocket
连接池管理

3.5 路由缓存

支持路由过滤器缓存，提高性能：

private final ConcurrentHashMap<Route, List<GatewayFilter>> routeFilterMap = new ConcurrentHashMap();  protected List<GatewayFilter> getCombinedFilters(Route route) {     if (this.routeFilterCacheEnabled) {         return routeFilterMap.computeIfAbsent(route, this::getAllFilters);     }     return getAllFilters(route); }