













请关注微信公众号:阿呆-bot
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 })
HandlerMapping(RoutePredicateHandlerMapping)@ConditionalOnClass(DispatcherHandler.class)
DispatcherHandler 是 WebFlux 的核心组件,类似于 WebMVC 的 DispatcherServlet@ConditionalOnProperty
# 可以通过配置控制是否启用
spring:
cloud:
gateway:
server:
webflux:
enabled: true # 默认启用,可以设置为 false 禁用
配置的核心组件:
RouteLocator:路由定位器,负责提供路由列表FilteringWebHandler:过滤器处理器,负责执行过滤器链RoutePredicateHandlerMapping:路由匹配器,负责匹配请求到路由GatewayProperties:Gateway 配置属性
这是 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 的属性中这是 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(); // 所有过滤器执行完毕
});
}
}
路由定位器,负责提供路由列表。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);

路由匹配是 Gateway 的核心功能。让我们用一个实际例子来说明整个流程:
场景:客户端请求 GET /api/users/123
详细流程:
DispatcherHandler 接收 HTTP 请求
客户端 → DispatcherHandler → Gateway
DispatcherHandler 是 WebFlux 的请求分发器,类似于 WebMVC 的 DispatcherServletHandlerMapping 处理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 异步匹配,支持异步条件判断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-serviceFilteringWebHandler,目标 URI 是 http://user-service:8081过滤器链是 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
-> 返回给客户端
关键点:
NettyRoutingFilter 是最后一个过滤器,负责实际转发请求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);
});
关键点:
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);
});
资源利用率高,少量线程处理大量请求
场景:处理 10000 并发请求
WebMVC: 需要 500 线程,内存 ~500MB
WebFlux: 只需要 16 线程,内存 ~100MB
资源利用率提升 5 倍!
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 # 路由过滤器
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); // 继续执行
}
路由匹配使用异步谓词,支持异步条件判断:
public interface AsyncPredicate<T> {
Mono<Boolean> test(T t);
}
支持的谓词:
使用 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(...);
}
}
特性:
支持路由过滤器缓存,提高性能:
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);
}
此内容由惯性聚合(RSS阅读器)自动聚合整理,仅供阅读参考。 原文来自 — 版权归原作者所有。