微服务组件源码6——Spring Gateway
Spring WebFlux 介绍
概述
* Spring WebFlux的定位和SpringMVC一样，不过区别在其处理方法可以使用Flux进行编写，且返回一个发布者Flux/Mono
* Spring WebFlux也可以适配@RequestMapping那一套的编写模式，只不过建议基于Flux来编写controller方法
WebFlux是兼容Spring MVC 基于@Controller，@RequestMapping等注解的编程开发方式的，可以做到平滑切换
WebFluxAutoConfiguration自动配置
* WebFluxAutoConfiguration 自动装配时先自动装配EnableWebFluxConfiguration 而EnableWebFluxConfiguration->DelegatingWebFluxConfiguration ->WebFluxConfigurationSupport。
* WebFluxConfigurationSupport 默认配置：
- webHandler底层实现：DispatcherHandler
- 同时配置了其他很多WebFlux核心组件包括 ：(这些组件的实现原理和Spring MVC很类似，只不过是基于Flux/Mono形式开发)
映射处理器处理器HandlerMapping和请求适配器HandlerAdapter：优先级为：RouterFunctionMapping、RequestMappingHandlerMapping、SimpleUrlHandlerMapping
异常处理器WebExceptionHandler：WebFluxResponseStatusExceptionHandler
响应处理器HandlerResultHandler : 优先级为：ResponseEntityResultHandler、ServerResponseResultHandler、ResponseBodyResultHandler、ViewResolutionResultHandler
@Configuration//条件装配 只有启动的类型是REACTIVE时加载
@ConditionalOnWebApplication(type = ConditionalOnWebApplication.Type.REACTIVE)//只有存在 WebFluxConfigurer实例 时加载
@ConditionalOnClass(WebFluxConfigurer.class)//在不存在 WebFluxConfigurationSupport实例时 加载
@ConditionalOnMissingBean({ WebFluxConfigurationSupport.class })//在之后装
@AutoConfigureAfter({ ReactiveWebServerFactoryAutoConfiguration.class,
CodecsAutoConfiguration.class, ValidationAutoConfiguration.class })//自动装配顺序
@AutoConfigureOrder(Ordered.HIGHEST_PRECEDENCE + 10)
public class WebFluxAutoConfiguration {
@Configuration
@EnableConfigurationProperties({ResourceProperties.class, WebFluxProperties.class})
//接口编程 在装配WebFluxConfig 之前要先 装配EnableWebFluxConfiguration
@Import({EnableWebFluxConfiguration.class})
public static class WebFluxConfig implements WebFluxConfigurer {
//隐藏部分源码 /** * Configuration equivalent to {@code @EnableWebFlux}. */
}
@Configuration
public static class EnableWebFluxConfiguration extends DelegatingWebFluxConfiguration {
//隐藏部分代码
public EnableWebFluxConfiguration(WebFluxProperties webFluxProperties,
ObjectProvider<WebFluxRegistrations> webFluxRegistrations) {
this.webFluxProperties = webFluxProperties;
this.webFluxRegistrations = webFluxRegistrations.getIfUnique();
}
//实现类DelegatingWebFluxConfiguration的创建RequestMappingHandlerAdapter和RequestMappingHandlerMapping的方法
//优先从容器里的WebFluxRegistrations来获取
@Override
protected RequestMappingHandlerAdapter createRequestMappingHandlerAdapter() {
if (this.webFluxRegistrations != null
&& this.webFluxRegistrations.getRequestMappingHandlerAdapter() != null) {
return this.webFluxRegistrations.getRequestMappingHandlerAdapter();
}
//默认为支持@RequestMapping注解的RequestMappingHandlerAdapter();
return super.createRequestMappingHandlerAdapter();
}
@Override
protected RequestMappingHandlerMapping createRequestMappingHandlerMapping() {
if (this.webFluxRegistrations != null
&& this.webFluxRegistrations.getRequestMappingHandlerMapping() != null) {
return this.webFluxRegistrations.getRequestMappingHandlerMapping();
}
//默认为支持@RequestMapping注解的RequestMappingHandlerMapping();
return super.createRequestMappingHandlerMapping();
}
}
@Configuration
@ConditionalOnEnabledResourceChain
static class ResourceChainCustomizerConfiguration {
//隐藏部分代码
//
}
private static class ResourceChainResourceHandlerRegistrationCustomizer
implements ResourceHandlerRegistrationCustomizer {
//隐藏部分代码
}
}
public class WebFluxConfigurationSupport implements ApplicationContextAware {
@Nullable
private Map<String, CorsConfiguration> corsConfigurations;
@Nullable
private PathMatchConfigurer pathMatchConfigurer;
@Nullable
private ViewResolverRegistry viewResolverRegistry;
@Nullable
private ApplicationContext applicationContext;
@Nullable
public final ApplicationContext getApplicationContext() {
return this.applicationContext;
}
@Override
public void setApplicationContext(@Nullable ApplicationContext applicationContext) {
this.applicationContext = applicationContext;
}
@Bean
public DispatcherHandler webHandler() {
return new DispatcherHandler();
}
@Bean
@Order(0)
public WebExceptionHandler responseStatusExceptionHandler() {
return new WebFluxResponseStatusExceptionHandler();
}
//默认为RequestMappingHandlerMapping
@Bean
public RequestMappingHandlerMapping requestMappingHandlerMapping(
@Qualifier("webFluxContentTypeResolver") RequestedContentTypeResolver contentTypeResolver) {
RequestMappingHandlerMapping mapping = createRequestMappingHandlerMapping();
mapping.setOrder(0);
mapping.setContentTypeResolver(contentTypeResolver);
mapping.setCorsConfigurations(getCorsConfigurations());
PathMatchConfigurer configurer = getPathMatchConfigurer();
Boolean useTrailingSlashMatch = configurer.isUseTrailingSlashMatch();
if (useTrailingSlashMatch != null) {
mapping.setUseTrailingSlashMatch(useTrailingSlashMatch);
}
Boolean useCaseSensitiveMatch = configurer.isUseCaseSensitiveMatch();
if (useCaseSensitiveMatch != null) {
mapping.setUseCaseSensitiveMatch(useCaseSensitiveMatch);
}
Map<String, Predicate<Class<?>>> pathPrefixes = configurer.getPathPrefixes();
if (pathPrefixes != null) {
mapping.setPathPrefixes(pathPrefixes);
}
return mapping;
}
protected RequestMappingHandlerMapping createRequestMappingHandlerMapping() {
return new RequestMappingHandlerMapping();
}
@Bean
public RouterFunctionMapping routerFunctionMapping(ServerCodecConfigurer serverCodecConfigurer) {
RouterFunctionMapping mapping = createRouterFunctionMapping();
mapping.setOrder(-1); // go before RequestMappingHandlerMapping
mapping.setMessageReaders(serverCodecConfigurer.getReaders());
mapping.setCorsConfigurations(getCorsConfigurations());
return mapping;
}
protected RouterFunctionMapping createRouterFunctionMapping() {
return new RouterFunctionMapping();
}
@Bean
public HandlerMapping resourceHandlerMapping(ResourceUrlProvider resourceUrlProvider) {
ResourceLoader resourceLoader = this.applicationContext;
if (resourceLoader == null) {
resourceLoader = new DefaultResourceLoader();
}
ResourceHandlerRegistry registry = new ResourceHandlerRegistry(resourceLoader);
registry.setResourceUrlProvider(resourceUrlProvider);
addResourceHandlers(registry);
AbstractHandlerMapping handlerMapping = registry.getHandlerMapping();
if (handlerMapping != null) {
PathMatchConfigurer configurer = getPathMatchConfigurer();
Boolean useTrailingSlashMatch = configurer.isUseTrailingSlashMatch();
Boolean useCaseSensitiveMatch = configurer.isUseCaseSensitiveMatch();
if (useTrailingSlashMatch != null) {
handlerMapping.setUseTrailingSlashMatch(useTrailingSlashMatch);
}
if (useCaseSensitiveMatch != null) {
handlerMapping.setUseCaseSensitiveMatch(useCaseSensitiveMatch);
}
}
else {
handlerMapping = new org.springframework.web.reactive.config.WebFluxConfigurationSupport.EmptyHandlerMapping();
}
return handlerMapping;
}
@Bean
public RequestMappingHandlerAdapter requestMappingHandlerAdapter(
@Qualifier("webFluxAdapterRegistry") ReactiveAdapterRegistry reactiveAdapterRegistry,
ServerCodecConfigurer serverCodecConfigurer,
@Qualifier("webFluxConversionService") FormattingConversionService conversionService,
@Qualifier("webFluxValidator") Validator validator) {
RequestMappingHandlerAdapter adapter = createRequestMappingHandlerAdapter();
adapter.setMessageReaders(serverCodecConfigurer.getReaders());
adapter.setWebBindingInitializer(getConfigurableWebBindingInitializer(conversionService, validator));
adapter.setReactiveAdapterRegistry(reactiveAdapterRegistry);
ArgumentResolverConfigurer configurer = new ArgumentResolverConfigurer();
configureArgumentResolvers(configurer);
adapter.setArgumentResolverConfigurer(configurer);
return adapter;
}
protected RequestMappingHandlerAdapter createRequestMappingHandlerAdapter() {
return new RequestMappingHandlerAdapter();
}
@Bean
public HandlerFunctionAdapter handlerFunctionAdapter() {
return new HandlerFunctionAdapter();
}
@Bean
public SimpleHandlerAdapter simpleHandlerAdapter() {
return new SimpleHandlerAdapter();
}
@Bean
public ResponseEntityResultHandler responseEntityResultHandler(
@Qualifier("webFluxAdapterRegistry") ReactiveAdapterRegistry reactiveAdapterRegistry,
ServerCodecConfigurer serverCodecConfigurer,
@Qualifier("webFluxContentTypeResolver") RequestedContentTypeResolver contentTypeResolver) {
return new ResponseEntityResultHandler(serverCodecConfigurer.getWriters(),
contentTypeResolver, reactiveAdapterRegistry);
}
@Bean
public ResponseBodyResultHandler responseBodyResultHandler(
@Qualifier("webFluxAdapterRegistry") ReactiveAdapterRegistry reactiveAdapterRegistry,
ServerCodecConfigurer serverCodecConfigurer,
@Qualifier("webFluxContentTypeResolver") RequestedContentTypeResolver contentTypeResolver) {
return new ResponseBodyResultHandler(serverCodecConfigurer.getWriters(),
contentTypeResolver, reactiveAdapterRegistry);
}
@Bean
public ViewResolutionResultHandler viewResolutionResultHandler(
@Qualifier("webFluxAdapterRegistry") ReactiveAdapterRegistry reactiveAdapterRegistry,
@Qualifier("webFluxContentTypeResolver") RequestedContentTypeResolver contentTypeResolver) {
ViewResolverRegistry registry = getViewResolverRegistry();
List<ViewResolver> resolvers = registry.getViewResolvers();
ViewResolutionResultHandler handler = new ViewResolutionResultHandler(
resolvers, contentTypeResolver, reactiveAdapterRegistry);
handler.setDefaultViews(registry.getDefaultViews());
handler.setOrder(registry.getOrder());
return handler;
}
@Bean
public ServerResponseResultHandler serverResponseResultHandler(
ServerCodecConfigurer serverCodecConfigurer) {
List<ViewResolver> resolvers = getViewResolverRegistry().getViewResolvers();
ServerResponseResultHandler handler = new ServerResponseResultHandler();
handler.setMessageWriters(serverCodecConfigurer.getWriters());
handler.setViewResolvers(resolvers);
return handler;
}
private static final class EmptyHandlerMapping extends AbstractHandlerMapping {
@Override
public Mono<Object> getHandlerInternal(ServerWebExchange exchange) {
return Mono.empty();
}
}
}
HTTPHandler
不同http服务器调用适配
* 使用一个简单的处理业务请求和响应的抽象，用来适配不同HTTP服务容器的API。
不同的HTTP服务容器在处理请求时需要调用此HTTPHandler，从而都会走到此HTTPHandler对象中
* 在HttpHandlerAutoConfiguration配置默认为HttpWebHandlerAdapter，里面装饰了一系列的WebHandler，以及实现Mono<Void> handle方法定义了请求响应的整个流程
* 例如在SpringBoot中使用ReactiveWebServerApplicationContext作用应用上下文。那么在容器刷新的时候会启动web服务，此时会基于不用的工厂创建不同的服务
//WebFlux 核心组件又是什么时候注入到对应的容器上下文中的呢？其实是在刷新容器上下文时注入进去的。
public class ReactiveWebServerApplicationContext extends GenericReactiveWebApplicationContext implements ConfigurableWebServerApplicationContext {
@Override
protected void onRefresh() {
super.onRefresh();
try {
createWebServer();
}
catch (Throwable ex) {
throw new ApplicationContextException("Unable to start reactive web server", ex);
}
}
private void createWebServer() {
WebServerManager serverManager = this.serverManager;
if (serverManager == null) {
String webServerFactoryBeanName = getWebServerFactoryBeanName();
ReactiveWebServerFactory webServerFactory = getWebServerFactory(webServerFactoryBeanName);
boolean lazyInit = getBeanFactory().getBeanDefinition(webServerFactoryBeanName).isLazyInit();
// 这里创建容器管理时注入httpHandler
this.serverManager = new WebServerManager(this, webServerFactory, this::getHttpHandler, lazyInit);
getBeanFactory().registerSingleton("webServerGracefulShutdown",
new WebServerGracefulShutdownLifecycle(this.serverManager));
// 注册一个 web容器启动服务类,该类继承了SmartLifecycle
getBeanFactory().registerSingleton("webServerStartStop",
new WebServerStartStopLifecycle(this.serverManager));
}
initPropertySources();
}
protected HttpHandler getHttpHandler() {
String[] beanNames = getBeanFactory().getBeanNamesForType(HttpHandler.class);
if (beanNames.length == 0) {
throw new ApplicationContextException(;
}
if (beanNames.length > 1) {
throw new ApplicationContextException();
}
//容器上下文获取httpHandler
return getBeanFactory().getBean(beanNames[0], HttpHandler.class);
}
}
* 例如NettyReactiveWebServerFactory工厂会创建Netty服务器，在其中的NettyServer里名就关联ReactorHttpHandlerAdapte(HttpWebHandlerAdapter)
SpringBoot提供的内嵌的NettyWebServer服务，基于NIO的模式，在请求到达的时候，就会让工作线程调用此ReactorHttpHandlerAdapte的apply方法
ReactorHttpHandlerAdapte的apply方法就是执行this.HttpWebHandlerAdapter.handle(request, response)，会使用DispachHander处理解析请求，定义好整个处理流程的Mono
接着Netty会直接订阅此Mono，触发整个流程
HttpWebHandlerAdapter实现类
* 在HttpHandlerAutoConfiguration配置里默认为HttpWebHandlerAdapter，里面装饰了一系列的WebHandler，以及实现Mono<Void> handle方法定义了请求响应的整个流程
- 在构建者Builder中会先收集好DispatcherHandler、FilteringWebHandler、WebExceptionHandler等信息，再进行builder
- 实现了HttpHandler接口的的handel方法
①创建ServerWebExchange，底层主要保存了 ServerHttpRequest request和 ServerHttpResponse response对象
②调用委托的装饰器处理类ExceptionHandlingWebHandler -> FilteringWebHandler -> DispatcherHandler 的handle方法 ---具体流程见下
- 如果最终发生异常，内部异常或者客户端连接异常返回Mono.empty()，其他返回Mono.error(ex);
- 如果调用没有异常，那么会执行response::setComplete做一些钩子方法的回调
@Override
public Mono<Void> handle(ServerHttpRequest request, ServerHttpResponse response) {
if (this.forwardedHeaderTransformer != null) {
request = this.forwardedHeaderTransformer.apply(request);
}
//创建ServerWebExchange，底层主要保存了 ServerHttpRequest request和 ServerHttpResponse response对象
ServerWebExchange exchange = createExchange(request, response);
//核心流程：
//1、调用委托的装饰器处理类ExceptionHandlingWebHandler -> FilteringWebHandler -> DispatcherHandler 的handle方法
//2、如果最终发生异常，内部异常或者客户端连接异常返回Mono.empty()，其他返回Mono.error(ex);
return getDelegate().handle(exchange)
.doOnSuccess(aVoid -> logResponse(exchange))
.onErrorResume(ex -> handleUnresolvedError(exchange, ex))
.then(Mono.defer(response::setComplete));
}
HttpHandlerAutoConfiguration自动配置HttpWebHandlerAdapter
* 其中从容器获取的WebHandler就是DispatcherHandler
WebHandler
WebHandler接口
* 一个用于处理业务请求抽象接口，定义了一系列处理行为
- 其中HTTPHandler实现类HttpWebHandlerAdapter也实现这个接口，底层依序装饰如下图的WebHandler
- 核心方法Mono<Void> handle(ServerWebExchange exchange);对请求进行处理，并返回一个Mono<Void>（基于Flux模式的开发）
public interface WebHandler {
Mono<Void> handle(ServerWebExchange exchange);
}
*相关核心实现类如下
DispatcherHandler
请求处理的总控制器，实际工作是由多个可配置的组件来处理。
请求分发处理器，Spring WebFlux 的访问入口。和 Spring MVC 的 DispatcherServlet 类似，主要流程如下
* 从容器中获取再排序List<HandlerMapping> handlerMappings、List<HandlerAdapter> handlerAdapters;、List<HandlerResultHandler> resultHandlers; 在自动配置类中默认的类型为
映射处理器处理器HandlerMapping和请求适配器HandlerAdapter：优先级为：RouterFunctionMapping、RequestMappingHandlerMapping、SimpleUrlHandlerMapping
* Flux.fromIterable(this.handlerMappings).concatMap(mapping -> mapping.getHandler(exchange))
遍历handlerMappings中收集到所有HandlerMapping实现类，调用其getHandler组装成新的Mono<Object>，在getHandler会定位到处理器
* flatMap(handler -> invokeHandler(exchange, handler))
对上一步返回的Mono<Object>，调用对应HandlerAdapter进行请求处理，返回Mono<HandlerResult>
使用handlerAdapter.supports(handler)，为true就return handlerAdapter.handle(exchange, handler);
* flatMap(result -> handleResult(exchange, result));
对上一步返回的Mono<HandlerResult>调用对应HandlerResultHandler 进行结果处理
使用resultHandler.supports(handlerResult)，为true就调用其handleResult(exchange, result)
public class DispatcherHandler implements WebHandler, ApplicationContextAware {
@Nullable
private List<HandlerMapping> handlerMappings;
@Nullable
private List<HandlerAdapter> handlerAdapters;
@Nullable
private List<HandlerResultHandler> resultHandlers;
public DispatcherHandler() {}
public DispatcherHandler(ApplicationContext applicationContext) {
initStrategies(applicationContext);
}
@Override
public void setApplicationContext(ApplicationContext applicationContext) {
initStrategies(applicationContext);
}
//从容器中获取在排序
protected void initStrategies(ApplicationContext context) {
Map<String, HandlerMapping> mappingBeans = BeanFactoryUtils.beansOfTypeIncludingAncestors(
context, HandlerMapping.class, true, false);
ArrayList<HandlerMapping> mappings = new ArrayList<>(mappingBeans.values());
AnnotationAwareOrderComparator.sort(mappings);
this.handlerMappings = ...;
this.resultHandlers = ...;
}
@Override
public Mono<Void> handle(ServerWebExchange exchange) {
if (this.handlerMappings == null) {
//Mono.error(ex);
return createNotFoundError();
}
//遍历一个执行handlerMappings中收集到HandlerMapping实现类，调用其getHandler组装成新的Flux，
return Flux.fromIterable(this.handlerMappings)
.concatMap(mapping -> mapping.getHandler(exchange))
//取一个元素执行的结果
.next()
.switchIfEmpty(createNotFoundError())
//再调用对应HandlerAdapter进行处理
.flatMap(handler -> invokeHandler(exchange, handler))
//再调用对应HandlerResultHandler 进行处理
.flatMap(result -> handleResult(exchange, result));
}
private Mono<HandlerResult> invokeHandler(ServerWebExchange exchange, Object handler) {
if (this.handlerAdapters != null) {
for (HandlerAdapter handlerAdapter : this.handlerAdapters) {
if (handlerAdapter.supports(handler)) {
return handlerAdapter.handle(exchange, handler);
}
}
}
return Mono.error(new IllegalStateException("No HandlerAdapter: " + handler));
}
private Mono<Void> handleResult(ServerWebExchange exchange, HandlerResult result) {
return getResultHandler(result).handleResult(exchange, result)
.checkpoint("Handler " + result.getHandler() + " [DispatcherHandler]")
.onErrorResume(ex ->
result.applyExceptionHandler(ex).flatMap(exResult -> {
String text = "Exception handler " + exResult.getHandler() +
", error=\"" + ex.getMessage() + "\" [DispatcherHandler]";
return getResultHandler(exResult).handleResult(exchange, exResult).checkpoint(text);
}));
}
private HandlerResultHandler getResultHandler(HandlerResult handlerResult) {
if (this.resultHandlers != null) {
for (HandlerResultHandler resultHandler : this.resultHandlers) {
if (resultHandler.supports(handlerResult)) {
return resultHandler;
}
}
}
throw new IllegalStateException("No HandlerResultHandler for " + handlerResult.getReturnValue());
}
}
FilteringWebHandler
* 底层封装了DispatcherHandler和拦截器
- 拦截器使用责任链模式，支持对DispatcherHandler进行前置后置的拦截
public class FilteringWebHandler extends WebHandlerDecorator {
private final DefaultWebFilterChain chain;
public FilteringWebHandler(WebHandler handler, List<WebFilter> filters) {
super(handler);
//handler表示DispatcherHandler，filters连接器集合
//每一个拦截器都会生成一个DefaultWebFilterChain，各个拦截器使用变量chain串联起来
//while (iterator.hasPrevious()) {
// chain = new DefaultWebFilterChain(filters, handler, iterator.previous(), chain);
// }
this.chain = new DefaultWebFilterChain(handler, filters);
}
@Override
public Mono<Void> handle(ServerWebExchange exchange) {
//返回拦截器链头结点的filter方法：如果存在链表可用，依序递归执行filter，否调用DispatcherHandler的handle
//在每一个具体的拦截器中，会自己具体是前置处理还是后置处理，是放行还是中断
// 返回：Mono.defer(() ->
// this.currentFilter != null && this.chain != null ?
// current.filter(exchange, chain):
// this.handler.handle(exchange));
return this.chain.filter(exchange);
}
}
//DefaultWebFilterChain：责任链设计：递归 + 拦截链表 + 在每一个具体的拦截器中，会自己具体是前置处理还是后置处理，是放行还是中断
@Override
public Mono<Void> filter(ServerWebExchange exchange) {
return Mono.defer(() ->
this.currentFilter != null && this.chain != null ?
invokeFilter(this.currentFilter, this.chain, exchange) :
this.handler.handle(exchange));
}
ExceptionHandlingWebHandler
* 先执行FilteringWebHandler的流程
* 在使用onErrorResume转换操作方法进封装为异常处理器发布者
- onErrorResume也是一个转换操作符，当发生异常时，会丢弃前面的流程，直接转为订阅到onErrorResume里指定的发布者
意味着当上链发生异常时，使用指定的exceptionHandlers一个一个处理，如果前面一个处理器接着抛出异常，那么就由第二个来处理。直到有一个成功执行就作为结果
public class ExceptionHandlingWebHandler extends WebHandlerDecorator {
private final List<WebExceptionHandler> exceptionHandlers;
public ExceptionHandlingWebHandler(WebHandler delegate, List<WebExceptionHandler> handlers) {
super(delegate);
List<WebExceptionHandler> handlersToUse = new ArrayList<>();
handlersToUse.add(new org.springframework.web.server.handler.ExceptionHandlingWebHandler.CheckpointInsertingHandler());
handlersToUse.addAll(handlers);
this.exceptionHandlers = Collections.unmodifiableList(handlersToUse);
}
@Override
public Mono<Void> handle(ServerWebExchange exchange) {
Mono<Void> completion;
//先执行FilteringWebHandler的流程
try {
completion = super.handle(exchange);
} catch (Throwable ex) {
completion = Mono.error(ex);
}
//在使用onErrorResume包装异常处理器
//onErrorResume也是一个转换操作符，当发生异常时，会丢弃前面的流程，直接转为订阅到onErrorResume里指定的发布者
//下面的流就是：使用指定的exceptionHandlers依序一个一个处理异常，如果前面一个处理器接着抛出异常，那么就由第二个来处理
//否则，异常处理的结果就会作为最终的处理结果发布者
for (WebExceptionHandler handler : this.exceptionHandlers) {
completion = completion.onErrorResume(ex -> handler.handle(exchange, ex));
}
return completion;
}
}
SpringCloud gateway整体流程
流程图
* 基于Spring webFlux的请求处理框架，DispatcherHandler首先会接受请求
- 使用内部注册的HandlerMappinp集合来匹配请求，这里优先选择RoutePredicateHandlerMapping来获取Object Handler处理器
- SimpleHandlerAdapter会适配处理到此将请求，直接调用Object Handler（FilteringWebHandler）处理器的handel方法，里面会使用predicate进行匹配到Route
* 此FilteringWebHandler包含指定的Route，其内也包含了内置和用户指定的Fliter
其中LoadBalancerClientFilter只用于把请求url进行负载均衡找出目标机器，在使用内部的httpClient发起远处调用
* 底层的http服务功能丰富，一般使用netty，因为netty的NIO支持和Flux的响应式编程适配得不错
RoutePredicateHandlerMapping
* SpringFlux中DispatchHandler处理请求的流程，这里就会优先使用RoutePredicateHandlerMapping，调用其getHandler->getHandlerInternal
Flux.fromIterable(this.handlerMappings)
.concatMap(mapping -> mapping.getHandler(exchange))
.next()
.switchIfEmpty(createNotFoundError())
.flatMap(handler -> invokeHandler(exchange, handler))
.flatMap(result -> handleResult(exchange, result));
* 父类AbstractHandlerMapping主要作用是调用子类的getHandlerInternal
public abstract class AbstractHandlerMapping extends ApplicationObjectSupport
implements HandlerMapping, Ordered, BeanNameAware {
//略
@Override
public Mono<Object> getHandler(ServerWebExchange exchange) {
return getHandlerInternal(exchange).map(handler -> {equest request = exchange.getRequest();
//略
return handler;
});
}
* RoutePredicateHandlerMapping 作用
①从其成员CachingRouteLocator中获取所有的Flux<Route>，遍历每一个Route。获取并执行PredicategetPredicate().apply(exchange)
②把第一个匹配成功的Route，存于请求属性中gatewayRoute -> Route,
③把RoutePredicateHandlerMapping中的FilteringWebHandler成员作为返回结果
public class RoutePredicateHandlerMapping extends AbstractHandlerMapping {
private final FilteringWebHandler webHandler;
private final RouteLocator routeLocator;
private final Integer managementPort;
private final RoutePredicateHandlerMapping.ManagementPortType managementPortType;
public RoutePredicateHandlerMapping(FilteringWebHandler webHandler, RouteLocator routeLocator, GlobalCorsProperties globalCorsProperties, Environment environment) {
this.webHandler = webHandler;
this.routeLocator = routeLocator;
this.managementPort = getPortProperty(environment, "management.server.");
this.managementPortType = this.getManagementPortType(environment);
this.setOrder(1);
this.setCorsConfigurations(globalCorsProperties.getCorsConfigurations());
}
protected Mono<?> getHandlerInternal(ServerWebExchange exchange) {
//如果managementPort管理端口，返回空
if (this.managementPortType == RoutePredicateHandlerMapping.ManagementPortType.DIFFERENT && this.managementPort != null && exchange.getRequest().getURI().getPort() == this.managementPort) {
return Mono.empty();
} else {
//调用lookupRoute，指定断言Predicate获取第一个匹配的Routes，
//存于请求属性中gatewayRoute -> Route, 返回RoutePredicateHandlerMapping中的FilteringWebHandler作为返回结果
exchange.getAttributes().put(ServerWebExchangeUtils.GATEWAY_HANDLER_MAPPER_ATTR, this.getSimpleName());
return this.lookupRoute(exchange).flatMap((r) -> {
exchange.getAttributes().remove(ServerWebExchangeUtils.GATEWAY_PREDICATE_ROUTE_ATTR);
exchange.getAttributes().put(ServerWebExchangeUtils.GATEWAY_ROUTE_ATTR, r);
return Mono.just(this.webHandler);
}).switchIfEmpty(Mono.empty().then(Mono.fromRunnable(() -> {
exchange.getAttributes().remove(ServerWebExchangeUtils.GATEWAY_PREDICATE_ROUTE_ATTR);
})));
}
}
//从其成员CachingRouteLocator中获取所有的Flux<Route>，遍历每一个Route。获取并执行PredicategetPredicate().apply(exchange)
//取第一个匹配成功的Route
protected Mono<Route> lookupRoute(ServerWebExchange exchange) {
return this.routeLocator.getRoutes().concatMap((route) -> {
return Mono.just(route).filterWhen((r) -> {
exchange.getAttributes().put(ServerWebExchangeUtils.GATEWAY_PREDICATE_ROUTE_ATTR, r.getId());
return (Publisher)r.getPredicate().apply(exchange);
}).onErrorResume((e) -> {
return Mono.empty();
});
}).next().map((route) -> {
this.validateRoute(route, exchange);
return route;
});
}
//...
}
FilteringWebHandler
* SpringFlux的处理器适配器SimpleHandlerAdapter会调用FilteringWebHandler的handle方法
* FilteringWebHandler的handle从请求属性中gatewayRoute获取route，获取其getFilters()全部配置的过滤器和全局过滤器，排序，在组装成DefaultGatewayFilterChain
* 过滤器链的底层List<GatewayFilter> filters作为链表，index表示当前执行的过滤器，
每一个具体的过滤器内会会调用DefaultGatewayFilterChain.filter，这样形成了递归调用，使得过滤器链的前后执行顺序是相反的
public class FilteringWebHandler implements WebHandler {
protected static final Log logger = LogFactory.getLog(FilteringWebHandler.class);
private final List<GatewayFilter> globalFilters;
//对传递进来的全局过滤器进行Order排序
public FilteringWebHandler(List<GlobalFilter> globalFilters) {
this.globalFilters = loadFilters(globalFilters);
}
private static List<GatewayFilter> loadFilters(List<GlobalFilter> filters) {
return filters.stream().map(filter -> {
//为一些没有实现filter(ServerWebExchange exchange, GatewayFilterChain chain) 的GlobalFilter进行适配一写
GatewayFilterAdapter gatewayFilter = new GatewayFilterAdapter(filter);
if (filter instanceof Ordered) {
int order = ((Ordered) filter).getOrder();
return new OrderedGatewayFilter(gatewayFilter, order);
}
return gatewayFilter;
}).collect(Collectors.toList());
}
//FilteringWebHandler的handle从请求属性中gatewayRoute获取route，获取其getFilters()全部配置的过滤器和全局过滤器，排序，在组装成DefaultGatewayFilterChain
@Override
public Mono<Void> handle(ServerWebExchange exchange) {
Route route = exchange.getRequiredAttribute(GATEWAY_ROUTE_ATTR);
List<GatewayFilter> gatewayFilters = route.getFilters();
List<GatewayFilter> combined = new ArrayList<>(this.globalFilters);
combined.addAll(gatewayFilters);
AnnotationAwareOrderComparator.sort(combined);
return new DefaultGatewayFilterChain(combined).filter(exchange);
}
//过滤器链
//底层List<GatewayFilter> filters作为链表，index表示当前执行的过滤器，
//每一个具体的过滤器内会会调用DefaultGatewayFilterChain.filter，这样形成了递归调用，使得过滤器链的前后执行顺序是相反的
private static class DefaultGatewayFilterChain implements GatewayFilterChain {
private final int index;
private final List<GatewayFilter> filters;
DefaultGatewayFilterChain(List<GatewayFilter> filters) {
this.filters = filters;
this.index = 0;
}
private DefaultGatewayFilterChain(DefaultGatewayFilterChain parent, int index) {
this.filters = parent.getFilters();
this.index = index;
}
public List<GatewayFilter> getFilters() {
return 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);
}
else {
return Mono.empty(); // complete
}
});
}
}
private static class GatewayFilterAdapter implements GatewayFilter {
private final GlobalFilter delegate;
GatewayFilterAdapter(GlobalFilter delegate) {
this.delegate = delegate;
}
@Override
public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
return this.delegate.filter(exchange, chain);
}
@Override
public String toString() {
final StringBuilder sb = new StringBuilder("GatewayFilterAdapter{");
sb.append("delegate=").append(delegate);
sb.append('}');
return sb.toString();
}
}
}
Gateway自动配置
自动配置类中可以看到Gateway的初始化过程
在 org.springframework.cloud.gateway.config 包下，我们可以看到四个配置类 ：
* GatewayAutoConfiguration
* GatewayClassPathWarningAutoConfiguration
* GatewayLoadBalancerClientAutoConfiguration
* GatewayRedisAutoConfiguration
GatewayClassPathWarningAutoConfiguration
Spring Cloud Gateway 2.x 基于 Spring WebFlux 实现。
GatewayClassPathWarningAutoConfiguration用于检查项目是否正确导入 spring-boot-starter-webflux 依赖，而不是错误导入 spring-boot-starter-web 依赖。
@Configuration(proxyBeanMethods = false)
@AutoConfigureBefore(GatewayAutoConfiguration.class)
@ConditionalOnProperty(name = "spring.cloud.gateway.enabled", matchIfMissing = true)
public class GatewayClassPathWarningAutoConfiguration {
private static final Log log = LogFactory
.getLog(GatewayClassPathWarningAutoConfiguration.class);
private static final String BORDER = "\n\n**********************************************************\n\n";
@Configuration(proxyBeanMethods = false)
@ConditionalOnClass(name = "org.springframework.web.servlet.DispatcherServlet")
@ConditionalOnWebApplication(type = ConditionalOnWebApplication.Type.SERVLET)
protected static class SpringMvcFoundOnClasspathConfiguration {
public SpringMvcFoundOnClasspathConfiguration() {
throw new MvcFoundOnClasspathException();
}
}
@Configuration(proxyBeanMethods = false)
@ConditionalOnMissingClass("org.springframework.web.reactive.DispatcherHandler")
protected static class WebfluxMissingFromClasspathConfiguration {
public WebfluxMissingFromClasspathConfiguration() {
log.warn(BORDER + "Spring Webflux is missing from the classpath, "
+ "which is required for Spring Cloud Gateway at this time. "
+ "Please add spring-boot-starter-webflux dependency." + BORDER);
}
}
}
GatewayLoadBalancerClientAutoConfiguration
* 初始化 LoadBalancerClientFilter过滤器，拦截请求进行负载均衡和服务发现
主要这时gateway自己定义的过滤器，不是Spring Flux的过滤器
* 需要从容器中获取LoadBalancerClient，例如RibbonLoadBalancerClient实现类
@Configuration(proxyBeanMethods = false)
@ConditionalOnClass({ LoadBalancerClient.class, RibbonAutoConfiguration.class,
DispatcherHandler.class })
@AutoConfigureAfter(RibbonAutoConfiguration.class)
@EnableConfigurationProperties(LoadBalancerProperties.class)
public class GatewayLoadBalancerClientAutoConfiguration {
@Bean
@ConditionalOnBean(LoadBalancerClient.class)
@ConditionalOnMissingBean({ LoadBalancerClientFilter.class,
ReactiveLoadBalancerClientFilter.class })
@ConditionalOnEnabledGlobalFilter
public LoadBalancerClientFilter loadBalancerClientFilter(LoadBalancerClient client,
LoadBalancerProperties properties) {
return new LoadBalancerClientFilter(client, properties);
}
}
GatewayRedisAutoConfiguration
GatewayRedisAutoConfiguration ，初始化 RedisRateLimiter 。
RequestRateLimiterGatewayFilterFactory 基于 RedisRateLimiter 实现网关的限流功能
GatewayAutoConfiguration（核心）
GatewayAutoConfiguration ，Spring Cloud Gateway 核心配置类，初始化如下的组件
NettyConfiguration
GlobalFilter
FilteringWebHandler
GatewayProperties
PrefixPathGatewayFilterFactory
RoutePredicateFactory
RouteDefinitionLocator
RouteLocator
RoutePredicateHandlerMapping
GatewayWebfluxEndpoint
* NettyConfiguration
①创建一个类型为 java.util.Objects.Consumer 的 Bean 对象。
该 Consumer 会将传入类型为 reactor.ipc.netty.options.HttpClientOptions.Builder 的参数 opts ，设置 opts 的 poolResources 属性。
调用 PoolResources.elastic("proxy")方法，创建name属性为 "proxy" 的reactor.ipc.netty.resources.PoolResources 。其中"proxy" 用于实际使用时，打印日志的标记。
②创建一个类型为HttpClient的Bean对象。该HttpClient使用Netty实现的Client 。
③创建一个类型为NettyRoutingFilter(HttpClient ②Bean)的Bean对象。
④创建一个类型为NettyWriteResponseFilter的Bean对象。
⑤创建一个类型为ReactorNettyWebSocketClient的Bean对象，用于下文 WebsocketRoutingFilter的Bean对象创建。
* GlobalFilter
①创建一个类型为RouteToRequestUrlFilter的Bean对象。
②创建一个类型为ForwardRoutingFilter的Bean对象
③创建一个类型为WebSocketService的Bean对象。
④创建一个类型为WebsocketRoutingFilter的Bean对象。
* FilteringWebHandler
- 当所有GlobalFilter 初始化完成时( 包括上面的 NettyRoutingFilter / NettyWriteResponseFilter )，创建一个类型为FilteringWebHandler(List<GlobalFilter> globalFilters) 的Bean对象
- 注意这时RoutePredicateHandlerMapping对象需要用到的FilteringWebHandler。和SpringFlux自己的FilteringWebHandler不一样
* GatewayProperties
创建一个类型为GatewayProperties 的Bean对象，用于加载配置文件配置的 RouteDefinition / FilterDefinition 。
* xxxFilterFactory
创建org.springframework.cloud.gateway.filter.factory 包下的 org.springframework.cloud.gateway.filter.factory.GatewayFilterFactory 接口的实现们
这些都是内置的过滤器
* xxxRoutePredicateFactory
创建org.springframework.cloud.gateway.handler.predicate 包下的 org.springframework.cloud.gateway.handler.predicate.RoutePredicateFactory 接口的实现们。
这些都是内置的predicate
* RouteDefinitionLocator
①创建一个RouteDefinitionLocator接口类型的PropertiesRouteDefinitionLocator(GatewayProperties Bean) 的Bean对象。
②创建一个类型为InMemoryRouteDefinitionRepository 的Bean对象。
③创建一个类型为CompositeRouteDefinitionLocator(List<RouteDefinitionLocator> routeDefinitionLocators①+②)的Bean对象。--@Primary
④基于DiscoveryClient注册发现的RouteDefinitionLocator实现类，需要手动引入配置，
* RouteLocator
①创建一个类型为RouteDefinitionRouteLocator的Bean对象
②创建一个类型为CachingRouteLocator(List<RouteLocator> routeLocators) 的Bean对象。该 Bean 对象内嵌 CompositeRouteLocator 对象。 --@Primary
* RoutePredicateHandlerMapping
- 创建一个类型为RoutePredicateHandlerMapping(FilteringWebHandler webHandler,RouteLocator routeLocator)的Bean对象，用于查找匹配到 Route ，并进行处理
- 是Spring webFlux需要的HandlerMapping接口实现类，其功能和SpringMVC中的一样，用于把一个请求映射为一个处理器。进行请求处理
其order为1，优先级位于RouterFunctionMapping之后
* GatewayWebfluxEndpoint
创建一个类型为 org.springframework.cloud.gateway.actuate.GatewayWebfluxEndpoint 的 Bean 对象，提供管理网关的 HTTP API
Route路由
整体流程
* RouteDefinitionLocator接口负责读取路由的配置，并生成配置RouteDefinition 集合发布者Flux<RouteDefinition> ，RouteDefinitionLocator 接口有四种实现 ：
①PropertiesRouteDefinitionLocator，从配置文件( 例如，YML / Properties 等 ) 读取。
②RouteDefinitionRepository，从存储器(例如，内存 / Redis / MySQL 等 )读取。
③DiscoveryClientRouteDefinitionLocator ，从注册中心( 例如，Eureka / Consul / Zookeeper / Etcd 等 )读取。
④CompositeRouteDefinitionLocator ，组合多种 RouteDefinitionLocator 的实现，为 RouteDefinitionRouteLocator 提供统一入口。
* RouteLocator接口负责生成路由的最终实体类Route，有如下几种方法
①通过 RouteDefinitionRouteLocator获取上一步的RouteDefinition，并转换成 Route 。
有多个实现类CompositeRouteLocator / CachingRouteLocator ，最重要的是RouteDefinitionRouteLocator
②可以通过代码定义自定义路由
public RouteLocator customRouteLocator() {
return Routes.locator()
.route("test")
.predicate(host("**.abc.org").and(path("/image/png")))
.addResponseHeader("X-TestHeader", "foobar")
.uri("http://httpbin.org:80")
.build();
}
RouteDefinitionLocator
概述
* RouteDefinitionLocator接口
负责读取路由的配置，并生成配置RouteDefinition 集合Flux<RouteDefinition> ，RouteDefinitionLocator 接口有四种实现 ：
①PropertiesRouteDefinitionLocator，从配置文件( 例如，YML / Properties 等 ) 读取。
②RouteDefinitionRepository，从存储器(例如，内存 / Redis / MySQL 等 )读取。
③DiscoveryClientRouteDefinitionLocator ，从注册中心( 例如，Eureka / Consul / Zookeeper / Etcd 等 )读取。
④CompositeRouteDefinitionLocator ，组合多种 RouteDefinitionLocator 的实现，为 RouteDefinitionRouteLocator 提供统一入口。
* 接口只有一个方法getRouteDefinitions以获取Flux<RouteDefinition>
public interface RouteDefinitionLocator {
Flux<RouteDefinition> getRouteDefinitions();
}
PropertiesRouteDefinitionLocator
* 从配置文件( 例如，YML / Properties 等 ) 读取路由配置
* 基于GatewayProperties，调用其getRoutes获取Flux<RouteDefinition>集合，Flux.fromIterable(this.properties.getRoutes());
* GatewayProperties借助@ConfigurationProperties(“spring.cloud.gateway”)，
①已经会配置文件的值映射到List<RouteDefinition> routes中
private List<RouteDefinition> routes = new ArrayList<>();
②添加一个默认的RemoveNonProxyHeadersGatewayFilterFactory到List<FilterDefinition>defaultFilters 默认富过滤器属性中
//GatewayAutoConfiguration
@Bean
@ConditionalOnMissingBean
public PropertiesRouteDefinitionLocator propertiesRouteDefinitionLocator(
GatewayProperties properties) {
return new PropertiesRouteDefinitionLocator(properties);
}
public class PropertiesRouteDefinitionLocator implements RouteDefinitionLocator {
private final GatewayProperties properties;
public PropertiesRouteDefinitionLocator(GatewayProperties properties) {
this.properties = properties;
}
@Override
public Flux<RouteDefinition> getRouteDefinitions() {
return Flux.fromIterable(this.properties.getRoutes());
}
}
@ConfigurationProperties("spring.cloud.gateway")
@Validated
public class GatewayProperties {
//ConfigurationProperties注解的作用已经会配置文件的值映射到此routes中
@NotNull
@Valid
private List<RouteDefinition> routes = new ArrayList<>();
//添加一个默认的List<FilterDefinition>定义
private List<FilterDefinition> defaultFilters = loadDefaults();
private ArrayList<FilterDefinition> loadDefaults() {
ArrayList<FilterDefinition> defaults = new ArrayList<>();
FilterDefinition definition = new FilterDefinition();
definition.setName(normalizeFilterName(RemoveNonProxyHeadersGatewayFilterFactory.class));
defaults.add(definition);
return defaults;
}
//略
}
* 配置文件举例
spring:
cloud:
gateway:
routes:
- id: payment_routh #payment_routh
uri: lb://cloud-provider-service #匹配后提供服务的路由地址，lb后跟提供服务的微服务的名
predicates:
- Path=/payment/get/** #断言，路径（是除了host之外的路径串）相匹配的进行路由，即把uri作为目标地址，
filters:
- AddRequestParameter=foo, bar #对匹配的请求，会额外添加foo=bar的请求参数。
- AddResponseHeader=X-Response-Foo, Bar #对匹配的请求，响应返回时会额外添加X-Response-Foo:Bar的header返回。
#如果存在=对，第一个会作为name,其他存于map中
spring:
cloud:
gateway:
routes:
- id: rewritepath_route
uri: lb://foo-server
predicates:
- Path=/foo/**
filters:
- RewritePath=/foo/(?<segment>.*), /$\{segment}
对于上面的例子，如果请求的路径是/foo/bar，则gateway会将请求路径改为/bar，在拼接到uri为lb://foo-server/bar
这个例子是常用的，如上foo是用于到此route，最终从服务列表中获取对应的最终ip+端口
DiscoveryClientRouteDefinitionLocator
通过调用 DiscoveryClient 获取注册在注册中心的服务列表，生成对应的 RouteDefinition 数组
* 获取对应的服务发现客户端，即注册中心的客户端对象DiscoveryClient
* 确定路由id的前缀routeIdPrefix，这里默认为DiscoveryClient 类名
* 遍历服务列表，处理每一个服务，生成Flux<RouteDefinition>
①设置 ID = routeIdPrefix_serviceId
②设置 URI = lb://serviceId
③添加 Path 匹配断言： -Path = /serviceId/**
④添加 Path 重写过滤器：- RewritePath=/serviceId/(?<segment>.*), /$\{segment}
public class DiscoveryClientRouteDefinitionLocator implements RouteDefinitionLocator {
private final DiscoveryClient discoveryClient;
private final String routeIdPrefix;
public DiscoveryClientRouteDefinitionLocator(DiscoveryClient discoveryClient) {
this.discoveryClient = discoveryClient;
this.routeIdPrefix = this.discoveryClient.getClass().getSimpleName() + "_";
}
@Override
public Flux<RouteDefinition> getRouteDefinitions() {
//遍历服务列表，处理每一个服务
return Flux.fromIterable(discoveryClient.getServices())
.map(serviceId -> {
RouteDefinition routeDefinition = new RouteDefinition();
// 设置 ID = routeIdPrefix_serviceId
routeDefinition.setId(this.routeIdPrefix + serviceId);
// 设置 URI = lb://serviceId
routeDefinition.setUri(URI.create("lb://" + serviceId));
// 添加 Path 匹配断言： -Path = /serviceId/**
PredicateDefinition subPredicate = new PredicateDefinition();
subPredicate.setName(normalizePredicateName(PathRoutePredicateFactory.class));
subPredicate.addArg(PATTERN_KEY, "/" + serviceId + "/**");
routeDefinition.getPredicates().add(subPredicate);
// 添加 Path 重写过滤器
// 添加过滤器：- - RewritePath=/serviceId/(?<segment>.*), /$\{segment}
FilterDefinition filter = new FilterDefinition();
filter.setName(normalizeFilterName(RewritePathGatewayFilterFactory.class));
String regex = "/" + serviceId + "/(?<remaining>.*)";
String replacement = "/${remaining}";
filter.addArg(REGEXP_KEY, regex);
filter.addArg(REPLACEMENT_KEY, replacement);
routeDefinition.getFilters().add(filter);
return routeDefinition;
});
}
}
CompositeRouteDefinitionLocator
* org.springframework.cloud.gateway.route.CompositeRouteDefinitionLocator ，组合多种 RouteDefinitionLocator 的实现，为 RouteDefinitionRouteLocator 提供统一入口。
* getRouteDefinitions() 方法，提供统一方法，将组合的 delegates 的路由定义全部返回。
delegates.flatMap会把底层保存的所有RouteDefinitionLocator实现类，把从每一个被观察者接受到的数据，调用getRouteDefinitions，全部组装成Flux<RouteDefinition>
代码如下 ：
public class CompositeRouteDefinitionLocator implements RouteDefinitionLocator {
/**
* RouteDefinitionLocator 数组
*/
private final Flux<RouteDefinitionLocator> delegates;
public CompositeRouteDefinitionLocator(Flux<RouteDefinitionLocator> delegates) {
this.delegates = delegates;
}
@Override
public Flux<RouteDefinition> getRouteDefinitions() {
return this.delegates.flatMap(RouteDefinitionLocator::getRouteDefinitions);
}
}
RouteDefinition
概述
* 通过RouteDefinitionLocator接口的实现类生成集合Flux<RouteDefinition>
* RouteDefinition只是一个路由的描述，此时还未生成对应的实体类，这个bean定义类似，包含了：
①id：ID 编号，唯一。
②List<PredicateDefinition> predicates：谓语定义数组。请求通过 predicates 判断是否匹配。在 Route 里，PredicateDefinition 转换成 Predicate 。
③List<FilterDefinition> filters：过滤器定义数组。在 Route 里，FilterDefinition 转换成 GatewayFilter 。
④uri：路由向的 URI 。
⑤order：顺序。当请求匹配到多个路由时，使用顺序小的。
* RouteDefinition 提供 text 字符串创建对象，按照一个规则
①text 参数，格式为 ${id}=${uri},${predicates[0]},${predicates[1]}...${predicates[n]} 。可以指定id和predicates，直接使用,隔开
举个例子, "route001=http://127.0.0.1,Host=**.addrequestparameter.org,Path=/get"
②filters和order属性，需要通过setter方法进行设置。
@Validated
public class RouteDefinition {
@NotEmpty
private String id = UUID.randomUUID().toString();
@NotEmpty
@Valid
private List<PredicateDefinition> predicates = new ArrayList<>();
@Valid
private List<FilterDefinition> filters = new ArrayList<>();
@NotNull
private URI uri;
private int order = 0;
public RouteDefinition() {
}
public RouteDefinition(String text) {
int eqIdx = text.indexOf('=');
if (eqIdx <= 0) {
throw new ValidationException("Unable to parse RouteDefinition text '" + text
+ "'" + ", must be of the form name=value");
}
setId(text.substring(0, eqIdx));
String[] args = tokenizeToStringArray(text.substring(eqIdx + 1), ",");
setUri(URI.create(args[0]));
for (int i = 1; i < args.length; i++) {
this.predicates.add(new PredicateDefinition(args[i]));
}
}
}
PredicateDefinition
* Predicate的定义，保存了如下信息：、
①name ：谓语定义名字。通过 name 对应到 org.springframework.cloud.gateway.handler.predicate.RoutePredicateFactory 的实现类。
例如说，name=Query 对应到 QueryRoutePredicateFactory 。
②args：参数数组。例如，name=Host 会转为args={"_genkey_0" : "iocoder.cn"} ，最终匹配请求的 hostname 为 iocoder.cn 。--其中_genkey_i是固定的模式
* 基于text 参数进行规则，格式为 ${name}=${args[0]},${args[1]}...${args[n]} 。
举个例子, "Host=iocoder.cn"
@Validated
public class PredicateDefinition {
@NotNull
private String name;
private Map<String, String> args = new LinkedHashMap<>();
public PredicateDefinition() {}
public PredicateDefinition(String text) {
int eqIdx = text.indexOf('=');
if (eqIdx <= 0) {
throw new ValidationException("Unable to parse PredicateDefinition text '"
+ text + "'" + ", must be of the form name=value");
}
setName(text.substring(0, eqIdx));
String[] args = tokenizeToStringArray(text.substring(eqIdx + 1), ",");
for (int i = 0; i < args.length; i++) {
this.args.put(NameUtils.generateName(i), args[i]);
}
}
//略
}
FilterDefinition
* 过滤器的定义，保存了如下信息：
①name：过滤器定义名字。通过 name 对应到 org.springframework.cloud.gateway.filter.factory.GatewayFilterFactory 的实现类。
例如说，name=AddRequestParameter 对应到 AddRequestParameterGatewayFilterFactory
②args：参数数组。例如，name=AddRequestParameter / args={"_genkey_0": "foo", "_genkey_1": "bar"} ，添加请求参数 foo 为 bar 。
* 基于text 字符串创建对象。格式为 ${name}=${args[0]},${args[1]}...${args[n]} 。
举个例子, "AddRequestParameter=foo, bar"
@Validated
public class FilterDefinition {
@NotNull
private String name;
private Map<String, String> args = new LinkedHashMap<>();
public FilterDefinition() {
}
public FilterDefinition(String text) {
int eqIdx = text.indexOf('=');
if (eqIdx <= 0) {
setName(text);
return;
}
setName(text.substring(0, eqIdx));
String[] args = tokenizeToStringArray(text.substring(eqIdx + 1), ",");
for (int i = 0; i < args.length; i++) {
this.args.put(NameUtils.generateName(i), args[i]);
}
}
//略
}
Route
org.springframework.cloud.gateway.route.Route ，路由。
* 对应了RouteDefinition定义，是真正的实体类，里面的PredicateDefinition和FilterDefinition统一会转为对应的实体类
* 可以通过Builder对Route 里属性进行设置，在最终构建
代码如下
public class Route implements Ordered {
/**
* 路由编号
*/
private final String id;
/**
* 路由向的 URI
*/
private final URI uri;
/**
* 顺序
*/
private final int order;
/**
* 谓语数组
*/
private final Predicate<ServerWebExchange> predicate;
/**
* 过滤器数组
*/
private final List<GatewayFilter> gatewayFilters;
private Route(String id, URI uri, int order,
AsyncPredicate<ServerWebExchange> predicate,
List<GatewayFilter> gatewayFilters, Map<String, Object> metadata) {
this.id = id;
this.uri = uri;
this.order = order;
this.predicate = predicate;
this.gatewayFilters = gatewayFilters;
this.metadata = metadata;
}
public static Builder builder() {
return new Builder();
}
public static Builder builder(RouteDefinition routeDefinition) {
// @formatter:off
return new Builder().id(routeDefinition.getId())
.uri(routeDefinition.getUri())
.order(routeDefinition.getOrder())
.metadata(routeDefinition.getMetadata());
// @formatter:on
}
//略
}
RouteLocator
概述
RouteLocator接口
负责生成路由的最终实体类Route，有如下几种方法
①通过 RouteDefinitionRouteLocator获取RouteDefinition，并转换成 Route 。
②可以通过代码定义自定义路由
public interface RouteLocator {
Flux<Route> getRoutes();
}
RouteDefinitionRouteLocator（核心）
@Bean
public RouteLocator routeDefinitionRouteLocator(GatewayProperties properties,
List<GatewayFilterFactory> GatewayFilters,
List<RoutePredicateFactory> predicates,
RouteDefinitionLocator routeDefinitionLocator,
@Qualifier("webFluxConversionService") ConversionService conversionService) {
return new RouteDefinitionRouteLocator(routeDefinitionLocator, predicates,
GatewayFilters, properties, conversionService);
}
主体流程
* 成员变量
①RouteDefinitionLocator routeDefinitionLocator
存储RouteDefinition的对象，比如PropertiesRouteDefinitionLocator从配置文件读取路由配置
②封装了ApplicationEventPublisher、;BeanFactory、Supplier<ConversionService> 、SpelExpressionParser、Supplier<Validator> validator;
主要作用是用于创建Predicate和Filter实现类中的内部Config对象，用于判断结果
③Map<String, RoutePredicateFactory> predicates
Predicate名称关联RoutePredicateFactory实现类，如 Before -> BeforeRoutePredicateFactory
④Map<String, GatewayFilterFactory> gatewayFilterFactories
Filter名称关联GatewayFilterFactory实现类，如AddRequestHeade -> AddRequestHeaderGatewayFilterFactory
⑤GatewayProperties gatewayProperties：配置类
* 构造方法内
①initFactories(predicates)
把传递进来的所有RoutePredicateFactory实现类，按照Predicate名称关联RoutePredicateFactory实现类规则，存储于Map<String, RoutePredicateFactory> predicates变量中
②把传递进来的所有GatewayFilterFactory实现类，按照Filter名称关联GatewayFilterFactory实现类规则，存储于Map<String, GatewayFilterFactory> gatewayFilterFactories变量中
* 实现getRoutes()核心方法获取Flux<Route>（发布者）
①从RouteDefinitionLocator获取Flux<RouteDefinition>再通过convertToRoute方法map操作转为Flux<Route>，这时惰性计算，转换还未开始
Flux<Route> routes = this.routeDefinitionLocator.getRouteDefinitions().map(this::convertToRoute);
②根据isFailOnRouteDefinitionError，决定如果其中转换出现错误，是否跳过，是的话最终返回的Flux<Route>的onErrorContinue（只打印日志，而不抛出异常）方法即可
* 核心的方法convertToRoute，把一个RouteDefinition转为Route
①AsyncPredicate<ServerWebExchange> predicate = combinePredicates(routeDefinition)，返回配置的Predicate，
- 从当前的routeDefinition中getPredicates()获取List<PredicateDefinition> predicates，即此有了配置的PredicateDefinition定义
- 先取第一个PredicateDefinition调用lookup(PredicateDefinition)方法：
- 通过Predicate名称从predicates 中获取具体的RoutePredicateFactory实现类，如BeforeRoutePredicateFactory
- 通过ConfigurationService(ConfigurationService)构建出RoutePredicateFactory类中内部类Config（流程见下），对应了用户配置的参数值
比如 - Before=2018-12-25T14:33:47.789+08:00。2018-12-25T14:33:47.789+08:00就会作为BeforeRoutePredicateFactory.Config的ZonedDateTime datetime变量中
- 使用RoutePredicateFactory.applyAsync，其内调用toAsyncPredicate(apply(config));
apply(config)是每一个xxxPredicateFactory实现类都会实现的抽象方法，返回Predicate<ServerWebExchange>，
例如BeforeRoutePredicateFactory的实现如下，使用GatewayPredicat类型对象返回，其内的test方法就是具体的Predicate判断
这样就组装了Predicate对象，里面用户配置信息Config和具体test判断GatewayPredicat
public Predicate<ServerWebExchange> apply(Config config) {
return new GatewayPredicate() {
@Override
public boolean test(ServerWebExchange serverWebExchange) {
final ZonedDateTime now = ZonedDateTime.now();
return now.isBefore(config.getDatetime());
}
};
}
d.把这个Predicate调用AsyncPredicate.from(predicate)形成DefaultAsyncPredicate（见下节）
- 剩下的PredicateDefinition定义一次调用lookup(PredicateDefinition)方法形成DefaultAsyncPredicate。然后第一个DefaultAsyncPredicate以AND的形成串联所有DefaultAsyncPredicate
返回最终的AsyncPredicate<ServerWebExchange>
②getFilters(routeDefinition)把当前routeDefinition中定义的过滤器和默认过滤器，调用loadGatewayFilters转换List<GatewayFilter>，遍历每一个FilterDefinition
- 通过Filter名称从gatewayFilterFactories 中获取具体的GatewayFilterFactory实现类，如AddRequestHeaderGatewayFilterFactory
- 通过ConfigurationService(ConfigurationService)构建出GatewayFilterFactory类中内部类Config，对应了用户配置的值（可能没有）
- 调用GatewayFilterFactory实现类的apply方法返回GatewayFilter，其内容就是过滤的增强处理
GatewayFilter的filter方法会返回Mono<Void>类型，
- 按照Ordered、编写顺序进行排序
public class AddRequestHeaderGatewayFilterFactory
extends AbstractNameValueGatewayFilterFactory {
@Override
public GatewayFilter apply(NameValueConfig config) {
return new GatewayFilter() {
@Override
public Mono<Void> filter(ServerWebExchange exchange,
GatewayFilterChain chain) {
String value = ServerWebExchangeUtils.expand(exchange, config.getValue());
ServerHttpRequest request = exchange.getRequest().mutate()
.header(config.getName(), value).build();
return chain.filter(exchange.mutate().request(request).build());
}
@Override
public String toString() {
return filterToStringCreator(AddRequestHeaderGatewayFilterFactory.this)
.append(config.getName(), config.getValue()).toString();
}
};
}
}
③封装为Route
new Route(this.id, this.uri, this.order, predicate,this.gatewayFilters, this.metadata);
public class RouteDefinitionRouteLocator implements RouteLocator, BeanFactoryAware, ApplicationEventPublisherAware {
public static final String DEFAULT_FILTERS = "defaultFilters";
private final RouteDefinitionLocator routeDefinitionLocator;
private final ConfigurationService configurationService;
private final Map<String, RoutePredicateFactory> predicates = new LinkedHashMap<>();
private final Map<String, GatewayFilterFactory> gatewayFilterFactories = new HashMap<>();
private final GatewayProperties gatewayProperties;
@Deprecated
public RouteDefinitionRouteLocator(RouteDefinitionLocator routeDefinitionLocator,
List<RoutePredicateFactory> predicates,
List<GatewayFilterFactory> gatewayFilterFactories,
GatewayProperties gatewayProperties, ConversionService conversionService) {
this.routeDefinitionLocator = routeDefinitionLocator;
this.configurationService = new ConfigurationService();
this.configurationService.setConversionService(conversionService);
initFactories(predicates);
gatewayFilterFactories.forEach(
factory -> this.gatewayFilterFactories.put(factory.name(), factory));
this.gatewayProperties = gatewayProperties;
}
public RouteDefinitionRouteLocator(RouteDefinitionLocator routeDefinitionLocator,
List<RoutePredicateFactory> predicates,
List<GatewayFilterFactory> gatewayFilterFactories,
GatewayProperties gatewayProperties,
ConfigurationService configurationService) {
this.routeDefinitionLocator = routeDefinitionLocator;
//封装了ApplicationEventPublisher publisher;BeanFactory beanFactory;Supplier<ConversionService> conversionService;
//SpelExpressionParser parser = new SpelExpressionParser();private Supplier<Validator> validator;
//主要作用是用于创建内部Config
this.configurationService = configurationService;
//例如BeforeRoutePredicateFactory，会在predicates变量中存储 Before -> BeforeRoutePredicateFactory
initFactories(predicates);
//同上
gatewayFilterFactories.forEach(
factory -> this.gatewayFilterFactories.put(factory.name(), factory));
this.gatewayProperties = gatewayProperties;
}
@Override
public Flux<Route> getRoutes() {
//从RouteDefinitionLocator获取Flux<RouteDefinition>再通过convertToRoute方法Map转为Flux<Route>
//这时惰性计算，转换还未开始
Flux<Route> routes = this.routeDefinitionLocator.getRouteDefinitions()
.map(this::convertToRoute);
//如果其中转换出现错误，是否跳过
if (!gatewayProperties.isFailOnRouteDefinitionError()) {
// instead of letting error bubble up, continue
routes = routes.onErrorContinue((error, obj) -> {
//值打印数据
}
});
}
return routes;
}
private Route convertToRoute(RouteDefinition routeDefinition) {
//Predicate定义转为AsyncPredicate<ServerWebExchange>
AsyncPredicate<ServerWebExchange> predicate = combinePredicates(routeDefinition);
//转换默认和定义的过滤器为具体的实现类。例如ResponseHeaderGatewayFilterFactory
List<GatewayFilter> gatewayFilters = getFilters(routeDefinition);
//new Route(this.id, this.uri, this.order, predicate,this.gatewayFilters, this.metadata);
return Route.async(routeDefinition).asyncPredicate(predicate)
.replaceFilters(gatewayFilters).build();
}
@SuppressWarnings("unchecked")
List<GatewayFilter> loadGatewayFilters(String id,
List<FilterDefinition> filterDefinitions) {
ArrayList<GatewayFilter> ordered = new ArrayList<>(filterDefinitions.size());
for (int i = 0; i < filterDefinitions.size(); i++) {
FilterDefinition definition = filterDefinitions.get(i);
GatewayFilterFactory factory = this.gatewayFilterFactories
.get(definition.getName());
if (factory == null) {
throw new IllegalArgumentException();
}
//获取配置的Filter内部类Config，
Object configuration = this.configurationService.with(factory)
.name(definition.getName())
.properties(definition.getArgs())
.eventFunction((bound, properties) -> new FilterArgsEvent(
RouteDefinitionRouteLocator.this, id, (Map<String, Object>) properties))
.bind();
if (configuration instanceof HasRouteId) {
HasRouteId hasRouteId = (HasRouteId) configuration;
hasRouteId.setRouteId(id);
}
//直接过滤器的apply返回一个GatewayFilter，其内的filter的方法就是具体的过滤器逻辑
GatewayFilter gatewayFilter = factory.apply(configuration);
if (gatewayFilter instanceof Ordered) {
ordered.add(gatewayFilter);
}
else {
ordered.add(new OrderedGatewayFilter(gatewayFilter, i + 1));
}
}
return ordered;
}
private List<GatewayFilter> getFilters(RouteDefinition routeDefinition) {
List<GatewayFilter> filters = new ArrayList<>();
//处理默认的过滤器
if (!this.gatewayProperties.getDefaultFilters().isEmpty()) {
filters.addAll(loadGatewayFilters(routeDefinition.getId(),
new ArrayList<>(this.gatewayProperties.getDefaultFilters())));
}
//处理调用的过滤器
if (!routeDefinition.getFilters().isEmpty()) {
filters.addAll(loadGatewayFilters(routeDefinition.getId(),
new ArrayList<>(routeDefinition.getFilters())));
}
AnnotationAwareOrderComparator.sort(filters);
return filters;
}
private AsyncPredicate<ServerWebExchange> combinePredicates(
RouteDefinition routeDefinition) {
List<PredicateDefinition> predicates = routeDefinition.getPredicates();
if (predicates == null || predicates.isEmpty()) {
return AsyncPredicate.from(exchange -> true);
}
//使用第一个PredicateDefinition生成Predicate
AsyncPredicate<ServerWebExchange> predicate = lookup(routeDefinition,predicates.get(0));
//剩下的PredicateDefinition一一生成Predicate在and
//形成AndAsyncPredicate，进行且的判断
// public Publisher<Boolean> apply(T t) {
// return Mono.from(left.apply(t)).flatMap(
// result -> !result ? Mono.just(false) : Mono.from(right.apply(t)));
for (PredicateDefinition andPredicate : predicates.subList(1,
predicates.size())) {
AsyncPredicate<ServerWebExchange> found = lookup(routeDefinition,
andPredicate);
predicate = predicate.and(found);
}
return predicate;
}
@SuppressWarnings("unchecked")
private AsyncPredicate<ServerWebExchange> lookup(RouteDefinition route,
PredicateDefinition predicate) {
//获取一个RoutePredicateFactory，如BeforeRoutePredicateFactory
RoutePredicateFactory<Object> factory = this.predicates.get(predicate.getName());
if (factory == null) {
throw new IllegalArgumentException();
}
//构建此说需要的配置Config内部类，如oBeforeRoutePredicateFactory.Config
Object config = this.configurationService.with(factory)
.name(predicate.getName())
.properties(predicate.getArgs())
.eventFunction((bound, properties) -> new PredicateArgsEvent(
RouteDefinitionRouteLocator.this, route.getId(), properties))
.bind();
// @formatter:on
//默认情况下为返回 DefaultAsyncPredicate，其里面delegat为Predicate接口实现类（如GatewayPredicate里面实现了test），
// 在DefaultAsyncPredicate的apply方法会返回一个Publisher<Boolean>，其里面使用Mono.just(delegate.test(t))
return factory.applyAsync(config);
}
@Override
@Deprecated
public void setBeanFactory/Validator/ApplicationEventPublisher(BeanFactory beanFactory) throws BeansException {
//略
}
}
内部config的实例化
* 在RouteDefinitionRouteLocator内把一个RouteDefinition转为Route时，进行了AsyncPredicate<ServerWebExchange>（Predicate实现类串联）的构建，其中Object config的实例流程如下
- 使用factory.newConfig()获取到指定的配置类，基于反射实现
- 使用ConfigurationUtils.bind把用户配置属性绑定到配置类中
//RouteDefinitionRouteLocator的convertToRoute
private AsyncPredicate<ServerWebExchange> lookup(RouteDefinition route,
PredicateDefinition predicate) {
RoutePredicateFactory<Object> factory = this.predicates.get(predicate.getName());
Map<String, String> args = predicate.getArgs();
Map<String, Object> properties = factory.shortcutType().normalize(args, factory,this.parser, this.beanFactory);
Object config = factory.newConfig();
ConfigurationUtils.bind(config, properties, factory.shortcutFieldPrefix(),predicate.getName(), validator, conversionService);
if (this.publisher != null) {
this.publisher.publishEvent(
new PredicateArgsEvent(this, route.getId(), properties));
}
return factory.applyAsync(config);
}
* RouteDefinitionRouteLocator主要借助ConfigurationService configurationService构建辅助类，with保存了具体的实现类，name保存了predicate的名称，properties保存参数Map<String, String> args
例如配置文件内容
predicates:
- name: ReadBodyPredicateFactory #TokenRoutePredicateFactory
args: #代表Config的参数
headerName: 'kkk','nnn' #以Map<String, String>存储
存储于PredicateDefinition中：
name：TokenRoutePredicateFactory
args ：{"_genkey_0" : "kkk"}
{"_genkey_1" : "nnn"}
* bind方法开始构建，调用normalizeProperties对原始的Map<String, String> args进行转换normalizeProperties
①调用configurable.shortcutType()获取枚举ShortcutType，其中默认的DEFAULT
具体的Predicate实现类可以重写shortcutType() 指定不用的枚举类型
②DEFAULT作用是借助Predicate实现类的shortcutFieldOrder方法，这里指定了内部类属性字段的顺序，应用户进行配置时的顺序一致
把原始的原始的Map<String, String> arg作为Map<字段名称, 配置alue> 形式
* 最后一步借助Bindable实例化Config对象，并基于normalizedProperties绑定变量的值，把属性值绑定为具体的Predicate实现类中
//构建此说需要的配置Config内部类，如BeforeRoutePredicateFactory.Config
Object config = this.configurationService.with(factory)
.name(predicate.getName())
.properties(predicate.getArgs())
.eventFunction((bound, properties) -> new PredicateArgsEvent(
RouteDefinitionRouteLocator.this, route.getId(), properties))
.bind();
public T bind() {
validate();
if (this.normalizedProperties == null) {
this.normalizedProperties = normalizeProperties();
}
T bound = doBind();
if (this.eventFunction != null && this.service.publisher != null) {
ApplicationEvent applicationEvent = this.eventFunction.apply(bound,
this.normalizedProperties);
this.service.publisher.publishEvent(applicationEvent);
}
return bound;
}
@Override
protected Map<String, Object> normalizeProperties() {
if (this.service.beanFactory != null) {
return this.configurable.shortcutType().normalize(this.properties,
this.configurable, this.service.parser, this.service.beanFactory);
}
return super.normalizeProperties();
}
@Override
protected T doBind() {
Bindable<T> bindable = Bindable.of(this.configurable.getConfigClass());
T bound = bindOrCreate(bindable, this.normalizedProperties, //properties转换的
this.configurable.shortcutFieldPrefix(),
/* this.name, */this.service.validator.get(),//配置的validator
this.service.conversionService.get());//配置的类型转换conversionService
return bound;
}
AsyncPredicate断言链
* 在Route的创建阶段会把所有Predicate以and形式组装成AsyncPredicate<ServerWebExchange>
* 调用AsyncPredicate的apply会得到一个Publisher<Boolean>发布者
一订阅的话就会执行内部的以and形成组成的Predicate实现类集合，执行全部Predicate实现类的apply方法中test结果
* AsyncPredicate接口继承Function，主要默认实现了apply(T t)，用于返回一个发布者Publisher<Boolean>
public interface AsyncPredicate<T> extends Function<T, Publisher<Boolean>> {..}
@FunctionalInterface
public interface Function<T, R> {
R apply(T t);
....
}
* 接口定义了默认的实现类DefaultAsyncPredicate，调用AsyncPredicate.from(Predicate predicate)会把一个具体的predicate实现类会转为AsyncPredicate<ServerWebExchange>
static AsyncPredicate<ServerWebExchange> from(Predicate<? super ServerWebExchange> predicate) {
return new DefaultAsyncPredicate<>(GatewayPredicate.wrapIfNeeded(predicate));
}
①内部管理一个delegate，对应一个具体的Predicate工厂实现类，如BeforeRoutePredicateFactory
②实现了apply方法，以Mono.just(delegate.test(t))返回发布者。这个发布者是数据载体即时的发布者，即在订阅之前delegate.test(t)就已经执行
当用户调用DefaultAsyncPredicate.apply方法时，delegate.test(t)也被调用了，不过此时还没有订阅者
class DefaultAsyncPredicate<T> implements AsyncPredicate<T> {
private final Predicate<T> delegate;
public DefaultAsyncPredicate(Predicate<T> delegate) {this.delegate = delegate;}
@Override
public Publisher<Boolean> apply(T t) {
return Mono.just(delegate.test(t));
}
}
* 提供and(other AsyncPredicate)、not、or方法把当前的AsyncPredicate与其他的AsyncPredicate进与或非组合判断
①以and为例
default AsyncPredicate<T> and(AsyncPredicate<? super T> other) {
return new AndAsyncPredicate<>(this, other);//this代表调用者，即左边的AsyncPredicate，这样就形成了链式编程
}
②AndAsyncPredicate
使用flatMap，接受上一个发布者的结果，进行处理，如果为false，直接返回false，否则执行另一个AsyncPredicate的apply判断
class AndAsyncPredicate<T> implements AsyncPredicate<T> {
private final AsyncPredicate<? super T> left;
private final AsyncPredicate<? super T> right;
public AndAsyncPredicate(AsyncPredicate<? super T> left, AsyncPredicate<? super T> right) {
this.left = left;
this.right = right;
}
@Override
public Publisher<Boolean> apply(T t) {
return Mono.from(left.apply(t)).flatMap(
result -> !result ? Mono.just(false) : Mono.from(right.apply(t)));
}
}
CompositeRouteLocator
组合多种 RouteLocator 的实现类，为 RoutePredicateHandlerMapping 提供统一入口访问路由。代码如下 ：
public class CompositeRouteLocator implements RouteLocator {
private final Flux<RouteLocator> delegates;
public CompositeRouteLocator(Flux<RouteLocator> delegates) {
this.delegates = delegates;
}
//提供统一方法，将组合的 delegates 的路由全部返回。
@Override
public Flux<Route> getRoutes() {
return this.delegates.flatMap(RouteLocator::getRoutes);
}
}
CachingRouteLocator
在底层委托类的基础上，在提供缓存的功能
public class CachingRouteLocator implements RouteLocator {
private final RouteLocator delegate;
/**
* 路由缓存
*/
private final AtomicReference<List<Route>> cachedRoutes = new AtomicReference<>();
public CachingRouteLocator(RouteLocator delegate) {
this.delegate = delegate;
//通过collectRoutes方法，从委托类中获取最新的结果，保存到cachedRoutes中
this.cachedRoutes.compareAndSet(null, collectRoutes());
}
private List<Route> collectRoutes() {
List<Route> routes = this.delegate.getRoutes().collectList().block();
// 排序
AnnotationAwareOrderComparator.sort(routes);
return routes;
}
//接口实现：从缓存中获取即可
@Override
public Flux<Route> getRoutes() {
return Flux.fromIterable(this.cachedRoutes.get());
}
//监听 org.springframework.context.ApplicationEvent.RefreshRoutesEvent 事件，刷新缓存。
@EventListener(RefreshRoutesEvent.class)
void handleRefresh() {
refresh();
}
public Flux<Route> refresh() {
return Flux.fromIterable(this.cachedRoutes.getAndUpdate(
routes -> CachingRouteLocator.this.collectRoutes()));
}
}
RoutePredicateFactory
概述
* RoutePredicateFactory接口
①核心接口方法，Predicate<ServerWebExchange> apply(C config);表示Predicate的规则判断，
其中参数表示用户配置的信息，一般每一个RoutePredicateFactory实现类都有一个自己的config内部类，表示用户指定的配置信息
②name()，默认的获得 RoutePredicateFactory 的名字的方法。该方法截取类名前半段，例如 QueryRoutePredicateFactory 的结果为 Query
@FunctionalInterface
public interface RoutePredicateFactory {
String PATTERN_KEY = "pattern";
//核心接口方法，表示Predicate的规则判断
Predicate<ServerWebExchange> apply(C config);
//获得 RoutePredicateFactory 的名字。该方法截取类名前半段，例如 QueryRoutePredicateFactory 的结果为 Query
default String name() {
return NameUtils.normalizePredicateName(getClass());
}
//略
}
* 实现类
AbstractRoutePredicateFactory
* 主要继承AbstractConfigurable，用于实例化内部配置类
* configClass表示内部类的的类型，提供例如newConfig()方法借助BeanUtils反射子类自定义configClass并获取实例
public abstract class AbstractRoutePredicateFactory<C> extends AbstractConfigurable<C> implements RoutePredicateFactory<C> {
public AbstractRoutePredicateFactory(Class<C> configClass) {
super(configClass);
}
}
public abstract class AbstractConfigurable<C> implements Configurable<C> {
private Class<C> configClass;
protected AbstractConfigurable(Class<C> configClass) {
this.configClass = configClass;
}
public Class<C> getConfigClass() {
return configClass;
}
@Override
public C newConfig() {
return BeanUtils.instantiateClass(this.configClass);
}
@Override
public String toString() {
return new ToStringCreator(this).append("configClass", configClass).toString();
}
}
After/Before/BetweeenRoutePredicateFactory
* 使用
- After=2017-01-20T17:42:47.789-07:00[America/Denver]
- Before=2017-01-20T17:42:47.789-07:00[America/Denver]
- Betweeen=2017-01-20T17:42:47.789-07:00[America/Denver], 2017-01-21T17:42:47.789-07:00[America/Denver]
* 实现原理
①内部配置类的Config的ZonedDateTime datetime保存了用户指定的时机
②apply中：基于ZonedDateTime 时间工具类进行时间范围的判断
注：Before/BetweeenRoutePredicateFactory实现类似，略
public class AfterRoutePredicateFactory extends AbstractRoutePredicateFactory<AfterRoutePredicateFactory.Config> {
/**
* DateTime key.
*/
public static final String DATETIME_KEY = "datetime";
public AfterRoutePredicateFactory() {
super(Config.class);
}
@Override
public List<String> shortcutFieldOrder() {
return Collections.singletonList(DATETIME_KEY);
}
@Override
public Predicate<ServerWebExchange> apply(Config config) {
return new GatewayPredicate() {
@Override
public boolean test(ServerWebExchange serverWebExchange) {
final ZonedDateTime now = ZonedDateTime.now();
return now.isAfter(config.getDatetime());
}
@Override
public String toString() {
return String.format("After: %s", config.getDatetime());
}
};
}
public static class Config {
@NotNull
private ZonedDateTime datetime;
public ZonedDateTime getDatetime() {
return datetime;
}
public void setDatetime(ZonedDateTime datetime) {
this.datetime = datetime;
}
}
}
HeaderRoutePredicateFactory（Cookie/Host/Method/Query）
* 这些都是请求基本信息的判断，较为简单
ServerWebExchange.getRequest().getHeader/Cookie/Host/Method/ Query等，获取后与配置进行正则匹配
* 例如HeaderRoutePredicateFactory请求指定 Cookie 正则匹配指定值
- Header=X-Request-Id, \d+
public class HeaderRoutePredicateFactory
extends AbstractRoutePredicateFactory<HeaderRoutePredicateFactory.Config> {
public static final String HEADER_KEY = "header";
public static final String REGEXP_KEY = "regexp";
public HeaderRoutePredicateFactory() {
super(Config.class);
}
@Override
public List<String> shortcutFieldOrder() {
return Arrays.asList(HEADER_KEY, REGEXP_KEY);
}
@Override
public Predicate<ServerWebExchange> apply(Config config) {
boolean hasRegex = !StringUtils.isEmpty(config.regexp);
return new GatewayPredicate() {
@Override
public boolean test(ServerWebExchange exchange) {
List<String> values = exchange.getRequest().getHeaders().getOrDefault(config.header, Collections.emptyList());
if (values.isEmpty()) {return false;}
if (hasRegex) {
for (int i = 0; i < values.size(); i++) {
String value = values.get(i);
if (value.matches(config.regexp)) {
return true;
}
}
return false;
}
return true;
}
};
}
@Validated
public static class Config {
@NotEmpty
private String header;
private String regexp;
...
}
}
PathRoutePredicateFactory
* 请求 Path 进行正则的匹配指定值。
在存在多个匹配表达式是，可以指定当第一个不匹配时候，是否匹配第二个正则
* 注意路径参数的处理，如果存在路径参数，会保存在请求属性中，值为一个map<路径字段名，值>
path=/foo/123 <=> /foo/{segment} --> request请求属性{‘segment’,123}
public class PathRoutePredicateFactory
extends AbstractRoutePredicateFactory<org.springframework.cloud.gateway.handler.predicate.PathRoutePredicateFactory.Config> {
private static final String MATCH_OPTIONAL_TRAILING_SEPARATOR_KEY = "matchOptionalTrailingSeparator";
private PathPatternParser pathPatternParser = new PathPatternParser();
public PathRoutePredicateFactory() {
super(org.springframework.cloud.gateway.handler.predicate.PathRoutePredicateFactory.Config.class);
}
public void setPathPatternParser(PathPatternParser pathPatternParser) {
this.pathPatternParser = pathPatternParser;
}
@Override
public List<String> shortcutFieldOrder() {
return Arrays.asList("patterns", MATCH_OPTIONAL_TRAILING_SEPARATOR_KEY);
}
@Override
public ShortcutType shortcutType() {
return ShortcutType.GATHER_LIST_TAIL_FLAG;
}
@Override
public Predicate<ServerWebExchange> apply(org.springframework.cloud.gateway.handler.predicate.PathRoutePredicateFactory.Config config) {
final ArrayList<PathPattern> pathPatterns = new ArrayList<>();
synchronized (this.pathPatternParser) {
//指定匹配表达式集合，指定当第一个不匹配时候，是否匹配第二个正则
pathPatternParser.setMatchOptionalTrailingSeparator(
config.isMatchOptionalTrailingSeparator());
config.getPatterns().forEach(pattern -> {
PathPattern pathPattern = this.pathPatternParser.parse(pattern);
pathPatterns.add(pathPattern);
});
}
return new GatewayPredicate() {
@Override
public boolean test(ServerWebExchange exchange) {
PathContainer path = parsePath(exchange.getRequest().getURI().getRawPath());
PathPattern match = null;
for (int i = 0; i < pathPatterns.size(); i++) {
PathPattern pathPattern = pathPatterns.get(i);
if (pathPattern.matches(path)) {
match = pathPattern;
break;
}
}
if (match != null) {
//如果存在路径参数，保存在请求属性中
//例如 path=/foo/123 <=> /foo/{segment}
PathMatchInfo pathMatchInfo = match.matchAndExtract(path);
putUriTemplateVariables(exchange, pathMatchInfo.getUriVariables());
return true;
}
else {
return false;
}
}
};
}
@Validated
public static class Config {
private List<String> patterns = new ArrayList<>();
private boolean matchOptionalTrailingSeparator = true;
@Deprecated
public String getPattern() {
if (!CollectionUtils.isEmpty(this.patterns)) {
return patterns.get(0);
}
return null;
}
...
}
}
RemoteAddrRoutePredicateFactory
*匹配请求来源 IP 在指定范围内。支持多个，支持网段
* 基于IpSubnetFilterRule 工具类实现
public class RemoteAddrRoutePredicateFactory
extends AbstractRoutePredicateFactory<RemoteAddrRoutePredicateFactory.Config> {
public RemoteAddrRoutePredicateFactory() {
super(Config.class);
}
@Override
public ShortcutType shortcutType() {
return GATHER_LIST;
}
@Override
public List<String> shortcutFieldOrder() {
return Collections.singletonList("sources");
}
@NotNull
private List<IpSubnetFilterRule> convert(List<String> values) {
List<IpSubnetFilterRule> sources = new ArrayList<>();
for (String arg : values) {
addSource(sources, arg);
}
return sources;
}
@Override
public Predicate<ServerWebExchange> apply(Config config) {
List<IpSubnetFilterRule> sources = convert(config.sources);
return new GatewayPredicate() {
@Override
public boolean test(ServerWebExchange exchange) {
InetSocketAddress remoteAddress = config.remoteAddressResolver
.resolve(exchange);
if (remoteAddress != null && remoteAddress.getAddress() != null) {
String hostAddress = remoteAddress.getAddress().getHostAddress();
String host = exchange.getRequest().getURI().getHost();
for (IpSubnetFilterRule source : sources) {
if (source.matches(remoteAddress)) {
return true;
}
}
}
return false;
}
};
}
private void addSource(List<IpSubnetFilterRule> sources, String source) {
if (!source.contains("/")) { // no netmask, add default
source = source + "/32";
}
String[] ipAddressCidrPrefix = source.split("/", 2);
String ipAddress = ipAddressCidrPrefix[0];
int cidrPrefix = Integer.parseInt(ipAddressCidrPrefix[1]);
sources.add(
new IpSubnetFilterRule(ipAddress, cidrPrefix, IpFilterRuleType.ACCEPT));
}
@Validated
public static class Config {
@NotEmpty
private List<String> sources = new ArrayList<>();
@NotNull
private RemoteAddressResolver remoteAddressResolver = new RemoteAddressResolver() {
};
public List<String> getSources() {
return sources;
}
public Config setSources(List<String> sources) {
this.sources = sources;
return this;
}
public Config setSources(String... sources) {
this.sources = Arrays.asList(sources);
return this;
}
public Config setRemoteAddressResolver(
RemoteAddressResolver remoteAddressResolver) {
this.remoteAddressResolver = remoteAddressResolver;
return this;
}
}
}
GatewayFilter
概述
* DispatcherHandler请求入口会在选定route路由之后，会把Route route = exchange.getRequiredAttribute(GATEWAY_ROUTE_ATTR)存于请求属性，且hanlderMappring返回了FilteringWebHandler
（注意这是gateway的同名对象，是处理器），接着SimpleHandlerAdapter会调用此FilteringWebHandler的andle(exchange)，会从请求属性中获取Route ，同上获取全局GlobalFilter
和此Route 的tGatewayFilte组装成DefaultGatewayFilterChain，开始执行
* 组装滤器链
FilteringWebHandler的handle从请求属性中gatewayRoute获取route，获取其getFilters()全部配置的过滤器GatewayFilter 和全局过滤器GlobalFilter，基于@Order排序，在组装成DefaultGatewayFilterChain
DefaultGatewayFilterChain.filter的方法开始执行每一个过滤器
*全局的GlobalFilter顺序，不是默认的，用户需要注入容器，
①NettyWriteResponseFilter：order(-1)，基于Netty的响应处理
②WebClientWriteResponseFilter：order(-1)，基于WebClient的响应处理
③RouteToRequestUrlFilter：order(10000)，根据匹配的 Route ，计算出请求的地址
...其他的GatewayFilterFactory 实现类：order(1~10000)
④LoadBalancerClientFilter：order(10100)，对目标服务器进行负载均衡
...其他的GatewayFilterFactory 实现类：order(10100~Integer.MAX_VALUE)
⑤ForwardRoutingFilter：order(Integer.MAX_VALUE)，路由内的重定向处理
⑥NettyRoutingFilter：order(Integer.MAX_VALUE)，其根据 http:// 或 https:// 前缀( Scheme )过滤处理，使用基于 Netty 实现的 HttpClient 请求后端 Http 服务。
⑦WebClientHttpRoutingFilter：order(Integer.MAX_VALUE)，其根据 ws:// / wss:// 前缀( Scheme )过滤处理，代理后端 Websocket 服务
⑧WebsocketRoutingFilter：order(Integer.MAX_VALUE)，其根据 http:// 或 https:// 前缀( Scheme )过滤处理，使用基于 org.springframework.cloud.gateway.filter.WebClient 实现的 HttpClient 请求后端 Http 服务。
* 在Spring Cloud默认注入这些过滤器：
org.springframework.cloud.gateway.filter.AdaptCachedBodyGlobalFilter;
org.springframework.cloud.gateway.filter.ForwardPathFilter;
org.springframework.cloud.gateway.filter.ForwardRoutingFilter;
org.springframework.cloud.gateway.filter.GlobalFilter;
org.springframework.cloud.gateway.filter.NettyRoutingFilter;
org.springframework.cloud.gateway.filter.NettyWriteResponseFilter;
org.springframework.cloud.gateway.filter.RemoveCachedBodyFilter;
org.springframework.cloud.gateway.filter.RouteToRequestUrlFilter;
org.springframework.cloud.gateway.filter.WebsocketRoutingFilter;
org.springframework.cloud.gateway.filter.WeightCalculatorWebFilter;
org.springframework.cloud.gateway.filter.LoadBalancerClientFilter
GatewayFilterFactory 的实现类
请求Header和Parameter相关
* AddRequestHeaderGatewayFilterFactory
添加指定请求Header为指定值。
- AddRequestHeader=X-Request-Foo, Bar
* RemoveRequestHeaderGatewayFilterFactory
移除指定请求Header为指定值。
* AddResponseHeaderGatewayFilterFactory
添加指定需要Header 为指定值。
*RemoveResponseHeaderGatewayFilterFactory
移除指定需要Header 为指定值。
* SetResponseHeaderGatewayFilterFactory
*RemoveNonProxyHeadersGatewayFilterFactory
移除请求 Proxy 相关的 Header 。默认值为 [ "Connection", "Keep-Alive", "Proxy-Authenticate", "Proxy-Authorization", "TE", "Trailer", "Transfer-Encoding", "Upgrade" ]
可以通过 spring.cloud.gateway.filter.remove-non-proxy-headers 配置
* AddRequestParameterGatewayFilterFactory
添加url查询参数
* 基本实现
①Tuple的功能和Predicate的内部配置类Config处理类似
②规则，基于exchange.getRequest().mutate()进行Header 的操作，会基于原ServerHttpRequest 创建创建新的 ServerWebExchange
public class AddRequestHeaderGatewayFilterFactory implements GatewayFilterFactory {
@Override
public List<String> argNames() {
return Arrays.asList(NAME_KEY, VALUE_KEY);
}
@Override
public GatewayFilter apply(Tuple args) {
String name = args.getString(NAME_KEY);
String value = args.getString(VALUE_KEY);
return (exchange, chain) -> { // GatewayFilter
// 创建新的 ServerHttpRequest
ServerHttpRequest request = exchange.getRequest().mutate()
.header(name, value)
.build();
// 创建新的 ServerWebExchange ，提交过滤器链继续过滤
return chain.filter(exchange.mutate().request(request).build());
};
}
//Config 略
}
Path相关
* RewritePathGatewayFilterFactory
根据配置的正则表达式 regexp ，使用配置的 replacement 重写请求 Path 。从功能目的上类似 《Module ngx_http_rewrite_module》 。
- RewritePath=/foo/(?<segment>.*), /$\{segment} ，$\ 用于替代 $ ，避免和 YAML 语法冲突。
* PrefixPathGatewayFilterFactory，
这个类只需要配置一个prefix参数，它可以给请求的URI添加prefix前缀
* SetPathGatewayFilterFactory
它提供了一种通过允许模板化路径段来操作请求路径的简单方法。 这使用了Spring Framework中的uri模板。 允许多个匹配的段。
predicates:
- Path=/foo/{segment}
filters:
- SetPath=/{segment}
对于/ foo / bar的请求路径，这将在发出下游请求之前将路径设置为/ bar。
public class RewritePathGatewayFilterFactory implements GatewayFilterFactory {
public static final String REGEXP_KEY = "regexp";
public static final String REPLACEMENT_KEY = "replacement";
@Override
public List<String> argNames() {
return Arrays.asList(REGEXP_KEY, REPLACEMENT_KEY);
}
@Override
public GatewayFilter apply(Tuple args) {
final String regex = args.getString(REGEXP_KEY);
// `$\` 用于替代 `$` ，避免和 YAML 语法冲突。
String replacement = args.getString(REPLACEMENT_KEY).replace("$\\", "$");
return (exchange, chain) -> {
ServerHttpRequest req = exchange.getRequest();
// 添加 原始请求URI 到 GATEWAY_ORIGINAL_REQUEST_URL_ATTR
addOriginalRequestUrl(exchange, req.getURI());
// 重写 Path
String path = req.getURI().getPath();
String newPath = path.replaceAll(regex, replacement);
// 创建新的 ServerHttpRequest
ServerHttpRequest request = req.mutate()
.path(newPath) // 设置 Path
.build();
// 添加 请求URI 到 GATEWAY_REQUEST_URL_ATTR
exchange.getAttributes().put(GATEWAY_REQUEST_URL_ATTR, request.getURI());
// 创建新的 ServerWebExchange ，提交过滤器链继续过滤
return chain.filter(exchange.mutate().request(request).build());
};
}
}
HystrixGatewayFilterFactory
* 使用例子，gateway降级后就会将请求转发到http://localhost:9994。
①gateway配置
spring:
cloud:
gateway:
routes:
- id: ingredients
uri: lb://ingredients
predicates:
- Path=//ingredients/**
filters:
- name: Hystrix
args:
name: fetchIngredients
fallbackUri: forward:/fallback
- id: ingredients-fallback
uri: http://localhost:9994
predicates:
- Path=/fallback
②Hystrix配置
hystrix.command.fetchIngredients.execution.isolation.strategy=...
* 基于 Hystrix ，实现基于 Route 级别的熔断功能
* 流程如下
①使用Setter：配置下HystrixCommand名字取用户配置的name，groupKey为当前过滤器类名
②基于RouteHystrixCommand实现命令的核心执行，继承HystrixObservableCommand，和HystrixCommand区别在于支持hot观察者模式、cold冷观察者模式，其整体逻辑和HystrixCommandl类似,
RouteHystrixCommand主要重写了HystrixObservableCommand的resumeWithFallback以支持对指定的fallbackUri进行DispatcherHandler请求入口的重新处理（对转发的地址进行重写路由）
③需要注意的是：
- 目前 Hystrix Command 执行超时时，返回客户端 504 状态码，如果使用 JSON 格式作为数据返回，则需要修改下该 HystrixGatewayFilter 的代码实现。
- 当Hystrix熔断时，最终返回客户端 200 状态码，内容为空，此处建议该 HystrixGatewayFilter 的代码实现。
public class HystrixGatewayFilterFactory implements GatewayFilterFactory {
@Override
public List<String> argNames() {
return Arrays.asList(NAME_KEY);
}
@Override
public GatewayFilter apply(Tuple args) {
//使用Setter配置下HystrixCommand名字取用户配置的name，groupKey为当前过滤器类名
final String commandName = args.getString(NAME_KEY);
final HystrixCommandGroupKey groupKey = HystrixCommandGroupKey.Factory.asKey(getClass().getSimpleName());
final HystrixCommandKey commandKey = HystrixCommandKey.Factory.asKey(commandName);
final HystrixObservableCommand.Setter setter = HystrixObservableCommand.Setter
.withGroupKey(groupKey)
.andCommandKey(commandKey);
//基于RouteHystrixCommand实现命令的核心执行，继承HystrixObservableCommand，和HystrixCommand区别在于支持hot观察者模式、cold冷观察者模式
//其整体逻辑和HystrixCommandl类似,主要重写了resumeWithFallback以支持对指定的fallbackUri进行DispatcherHandler请求入口的重新处理
return (exchange, chain) -> {
RouteHystrixCommand command = new RouteHystrixCommand(setter, exchange, chain);
return Mono.create(s -> {
// 使用 Hystrix Command Observable 订阅
Subscription sub = command.toObservable().subscribe(s::success, s::error, s::success);
// Mono 取消时，取消 Hystrix Command Observable 的订阅，结束 Hystrix Command 的执行
s.onCancel(sub::unsubscribe);
}).onErrorResume((Function<Throwable, Mono<Void>>) throwable -> {
//当 Hystrix Command 执行超时时，设置响应 504 状态码，并回写客户端响应( exchange.getResponse().setComplete() ) 。
if (throwable instanceof HystrixRuntimeException) {
HystrixRuntimeException e = (HystrixRuntimeException) throwable;
if (e.getFailureType() == TIMEOUT) { //TODO: optionally set status
setResponseStatus(exchange, HttpStatus.GATEWAY_TIMEOUT);
return exchange.getResponse().setComplete();
}
}
//当 Hystrix Command 发生其他异常时，例如断路器打开，返回 Mono.empty() ，最终返回客户端 200 状态码，内容为空。
return Mono.empty();
//调用 Mono#then() 方法，参数为空，返回空 Mono ，不再向后发射数据。
}).then();
};
}
}
RouteToRequestUrlFilter
* 根据匹配的 Route ，计算请求的地址。使用Route的url属性，这个会形成uri
* 把原请求的uri的各个部分重写为指定的Route的uri。Route指定的uri没有指定的部分，原请求的uri不会变
* 设置 requestUrl 到 GATEWAY_REQUEST_URL_ATTR 请求属性中
* 注意；如果 Route.uri 属性配置带有 Path ，则会覆盖请求的 Path 。
因为当原Request的uri参数有Path时，新建一个 CompositePathComponentBuilder赋值为原Request的的，因此原有Route.uri被覆盖了无效了
Route.uri：bin.org:80/123
原Request.uri：http://127.0.0.1:8080/test/segment
最终requestUrl：httpbin.org:80/test/segmen…
public class RouteToRequestUrlFilter implements GlobalFilter, Ordered {
private static final Log log = LogFactory.getLog(RouteToRequestUrlFilter.class);
public static final int ROUTE_TO_URL_FILTER_ORDER = 10000;
@Override
public int getOrder() {
return ROUTE_TO_URL_FILTER_ORDER;
}
@Override
public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
// 根据请求域属性GATEWAY_ROUTE_ATTR获得之前匹配到的 Route
Route route = exchange.getAttribute(GATEWAY_ROUTE_ATTR);
if (route == null) {
return chain.filter(exchange);
}
// 拼接 requestUrl，把原请求的uri的各个部分重写为指定的Route的uri
Route指定的uri没有指定的部分，原请求的uri不会变
URI requestUrl = UriComponentsBuilder.fromHttpRequest(exchange.getRequest())
.uri(route.getUri())
.build(true) // encoded=true
.toUri();
// 设置 requestUrl 到 GATEWAY_REQUEST_URL_ATTR {@link RewritePathGatewayFilterFactory}
exchange.getAttributes().put(GATEWAY_REQUEST_URL_ATTR, requestUrl);
// 提交过滤器链继续过滤
return chain.filter(exchange);
}
}
public UriComponentsBuilder uri(URI uri) {
Assert.notNull(uri, "URI must not be null");
this.scheme = uri.getScheme();
if (uri.isOpaque()) {
this.ssp = uri.getRawSchemeSpecificPart();
resetHierarchicalComponents();
}
else {
if (uri.getRawUserInfo() != null) {
this.userInfo = uri.getRawUserInfo();
}
if (uri.getHost() != null) {
this.host = uri.getHost();
}
if (uri.getPort() != -1) {
this.port = String.valueOf(uri.getPort());
}
if (StringUtils.hasLength(uri.getRawPath())) {
this.pathBuilder = new CompositePathComponentBuilder();
this.pathBuilder.addPath(uri.getRawPath());
}
if (StringUtils.hasLength(uri.getRawQuery())) {
this.queryParams.clear();
query(uri.getRawQuery());
}
resetSchemeSpecificPart();
}
if (uri.getRawFragment() != null) {
this.fragment = uri.getRawFragment();
}
return this;
}
ForwardRoutingFilter
spring:
application:
name: juejin-gateway
cloud:
gateway:
routes:
# =====================================
- id: forward_sample
uri: forward:///globalfilters
order: 10000
predicates:
- Path=/globalfilters
filters:
- PrefixPath=/application/gateway
* 处理流程
我们假定网关端口为 8080 ，当请求 http://127.0.0.1:8080/globalfilters 接口，Spring Cloud Gateway 处理过程如下 ：
①匹配到路由 Route (id = forward_sample) 。
②配置的PrefixPathGatewayFilterFactory 将请求改写成 http://127.0.0.1:8080/application/gateway/globalfilters 。
③ForwardRoutingFilter（顺序晚于PrefixPathGatewayFilterFactory ） 判断有 forward:// 前缀( Scheme )，过滤处理，将请求转发给 DispatcherHandler 重写走handle流程。
④DispatcherHandler 匹配并转发到当前网关实例本地接口 application/gateway/globalfilters 。
注：什么需要配置 PrefixPathGatewayFilterFactory ？需要通过 PrefixPathGatewayFilterFactory 将请求重写路径，以匹配本地 API ，否则 DispatcherHandler 转发会失败。
public class ForwardRoutingFilter implements GlobalFilter, Ordered {
private final DispatcherHandler dispatcherHandler;
public ForwardRoutingFilter(DispatcherHandler dispatcherHandler) {
this.dispatcherHandler = dispatcherHandler;
}
@Override
public int getOrder() {
return Ordered.LOWEST_PRECEDENCE;
}
@Override
public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
// 获得 requestUrl
URI requestUrl = exchange.getRequiredAttribute(GATEWAY_REQUEST_URL_ATTR);
// 判断是否能够处理：forward:// 前缀或者该请求暂未被其他 Routing 网关处理
String scheme = requestUrl.getScheme();
if (isAlreadyRouted(exchange) || !scheme.equals("forward")) {
return chain.filter(exchange);
}
// 设置已经路由，exchange.getAttributes().put(GATEWAY_ALREADY_ROUTED_ATTR, true);
setAlreadyRouted(exchange);
// DispatcherHandler 匹配并转发到当前网关实例本地接口
return this.dispatcherHandler.handle(exchange);
}
}
LoadBalancerClientFilter
根据 lb:// 前缀过滤处理，使用 serviceId 选择一个服务实例，从而实现负载均衡。
* 获得 URL 。只处理 lb:// 为前缀( Scheme )的地址。
* 调用 ServerWebExchangeUtils#addOriginalRequestUrl(...) 添加原始请求 URI 到 GATEWAY_ORIGINAL_REQUEST_URL_ATTR
* 调用 LoadBalancerClient#choose(String) 方法，获得一个服务实例( ServiceInstance ) ，从而实现负载均衡。
例如基于ribbon实现的负载均衡器，RibbonLoadBalancerClient实现类
* 获取到服务实例uri后，覆盖原请求uri的scheme、host、port即可
public class LoadBalancerClientFilter implements GlobalFilter, Ordered {
public static final int LOAD_BALANCER_CLIENT_FILTER_ORDER = 10100;
private final LoadBalancerClient loadBalancer;
public LoadBalancerClientFilter(LoadBalancerClient loadBalancer) {
this.loadBalancer = loadBalancer;
}
@Override
public int getOrder() {
return LOAD_BALANCER_CLIENT_FILTER_ORDER;
}
@Override
public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
// 获得 URL
URI url = exchange.getAttribute(GATEWAY_REQUEST_URL_ATTR);
if (url == null || !url.getScheme().equals("lb")) {
return chain.filter(exchange);
}
// 添加 原始请求URI 到 GATEWAY_ORIGINAL_REQUEST_URL_ATTR
//preserve the original url
addOriginalRequestUrl(exchange, url);
log.trace("LoadBalancerClientFilter url before: " + url);
// 获取 服务实例
final ServiceInstance instance = loadBalancer.choose(url.getHost());
if (instance == null) {
throw new NotFoundException("Unable to find instance for " + url.getHost());
}
/*URI uri = exchange.getRequest().getURI();
URI requestUrl = loadBalancer.reconstructURI(instance, uri);*/
URI requestUrl = UriComponentsBuilder.fromUri(url)
.scheme(instance.isSecure()? "https" : "http") //TODO: support websockets
.host(instance.getHost())
.port(instance.getPort())
.build(true)
.toUri();
log.trace("LoadBalancerClientFilter url chosen: " + requestUrl);
// 添加 请求URI 到 GATEWAY_REQUEST_URL_ATTR
exchange.getAttributes().put(GATEWAY_REQUEST_URL_ATTR, requestUrl);
// 提交过滤器链继续过滤
return chain.filter(exchange);
}
}
RequestRateLimiterGatewayFilterFactory
RequestRateLimiterGatewayFilter 使用 Redis + Lua 实现分布式限流。
而限流的粒度，例如 URL / 用户 / IP 等，通过 org.springframework.cloud.gateway.filter.ratelimit.KeyResolver 实现类决定
filters:
- RequestRateLimiter=10, 20, #{@remoteAddrKeyResolver}
自定义 KeyResolver
实现一个使用请求 IP 作为限流键的 KeyResolver
@Bean(name = RemoteAddrKeyResolver.BEAN_NAME)
@ConditionalOnBean(RateLimiter.class)
public RemoteAddrKeyResolver remoteAddrKeyResolver() {
return new RemoteAddrKeyResolver();
}
public class RemoteAddrKeyResolver implements KeyResolver {
public static final String BEAN_NAME = "remoteAddrKeyResolver";
@Override
public Mono<String> resolve(ServerWebExchange exchange) {
return Mono.just(exchange.getRequest().getRemoteAddress().getAddress().getHostAddress());
}
}
处理流程
* 配置文件中指定了参数
①replenishRate：qps
②burstCapacity：桶的上限
③keyResolver：token的维度，如ip。
* 调用rateLimiter.isAllowed进行限处理
rateLimiter 属性，限流器。默认情况下，使用 RedisRateLimiter，在自动配置类中可以自定义指定
* 基于redis的令牌算法实现
①Redis会为每一个keyResolver的值对应一个令牌桶，如ip1和ip2使用不同的桶，用户传入配置的keyResolver可以解析处理桶的id
②用户还需要根据配置，给redis传递qps、桶的上限，已经此桶id最后一次添加令牌的时机
注：不能在 Redis Lua中 使用 TIME 获取时间戳，因此只好从应用获取然后传入，在某些极端情况下（机器时钟不准的情况下），限流会存在一些小问题
③在redis基于LUA实现了令牌桶的算法流程（略）
与基本的令牌桶的算法不同的是，redis需要计算消耗整个令牌桶填充需要多久时间ttl ，单位：秒。最为桶在redis的过期时间
local fill_time = capacity/rate，即桶大小/qps
local ttl = math.floor(fill_time*2)，在乘2
redis.call("setex", tokens_key, ttl, new_tokens)
public class RequestRateLimiterGatewayFilterFactory implements GatewayFilterFactory {
public static final String KEY_RESOLVER_KEY = "keyResolver";
private final RateLimiter rateLimiter;
private final KeyResolver defaultKeyResolver;
public RequestRateLimiterGatewayFilterFactory(RateLimiter rateLimiter,
KeyResolver defaultKeyResolver) {
this.rateLimiter = rateLimiter;
this.defaultKeyResolver = defaultKeyResolver;
}
@Override
public List<String> argNames() {
return Arrays.asList(
RedisRateLimiter.REPLENISH_RATE_KEY,
RedisRateLimiter.BURST_CAPACITY_KEY,
KEY_RESOLVER_KEY
);
}
@Override
public boolean validateArgs() {
return false;
}
@SuppressWarnings("unchecked")
@Override
public GatewayFilter apply(Tuple args) {
validateMin(2, args);
// 获得 KeyResolver
KeyResolver keyResolver;
if (args.hasFieldName(KEY_RESOLVER_KEY)) {
keyResolver = args.getValue(KEY_RESOLVER_KEY, KeyResolver.class);
} else {
keyResolver = defaultKeyResolver;
}
return (exchange, chain) -> keyResolver.resolve(exchange).flatMap(key ->
// TODO: if key is empty?
rateLimiter.isAllowed(key, args).flatMap(response -> {
// TODO: set some headers for rate, tokens left
// 允许访问
if (response.isAllowed()) {
return chain.filter(exchange);
}
// 被限流，不允许访问
exchange.getResponse().setStatusCode(HttpStatus.TOO_MANY_REQUESTS);
return exchange.getResponse().setComplete();
}));
}
}
NettyRoutingFilter转发到后端
* 一般位于过滤器链的最后一个，负责访问最终url
* Netty 路由网关过滤器。其根据 http:// 或 https:// 前缀( Scheme )过滤处理，使用基于 Netty 实现的 HttpClient 请求后端 Http 服务
基于Netty实现的HttpClient,在自动配置中可以指定
* NettyWriteResponseFilter是与 NettyRoutingFilter 成对使用的网关过滤器。其将 NettyRoutingFilter 请求后端 Http 服务的响应写回客户端。
主要这里的两个不同的链路：原请求和响应、以及Netty 发起的请求和响应
* 处理流程
①获得 requestUrl
②判断是否能够处理
- http:// 或者 https:// 前缀( Scheme ) 。
- 调用 ServerWebExchangeUtils#isAlreadyRouted(ServerWebExchange) 方法，判断该请求暂未被其他 Routing 网关处理
③获取原请求对象，以获取Method、url、请求头组成httpHeaders（可以自定义过滤）
④开始发起请求
- 使用Netty实现的ttpClient
- 创建 Netty的Request 对象，调用send发送
⑤doOnNext成功发送数据时触发，res为Netty client的响应，这些处理是不会阻塞的，因为此时还没有读取转发请求的响应
- 获取原响应，将 Netty Response 的响应头和响应状态赋值给原响应
- 设置 Netty Response 到 CLIENT_RESPONSE_ATTR，最后NettyWriteResponseFilter会进行处理
public class NettyRoutingFilter implements GlobalFilter, Ordered {
//基于Netty实现的HttpClient,在自动配置中可以指定
private final HttpClient httpClient;
public NettyRoutingFilter(HttpClient httpClient) {
this.httpClient = httpClient;
}
//返回顺序为 Integer.MAX_VALUE 最后一个
@Override
public int getOrder() {
return Ordered.LOWEST_PRECEDENCE;
}
@Override
public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
// 获得 requestUrl
URI requestUrl = exchange.getRequiredAttribute(GATEWAY_REQUEST_URL_ATTR);
// 判断是否能够处理
// http:// 或者 https:// 前缀( Scheme ) 。
// 调用 ServerWebExchangeUtils#isAlreadyRouted(ServerWebExchange) 方法，判断该请求暂未被其他 Routing 网关处理
String scheme = requestUrl.getScheme();
if (isAlreadyRouted(exchange) || (!scheme.equals("http") && !scheme.equals("https"))) {
return chain.filter(exchange);
}
// 设置已经路由
setAlreadyRouted(exchange);
//获取原请求对象，以获取如下信息
ServerHttpRequest request = exchange.getRequest();
// Request Method
final HttpMethod method = HttpMethod.valueOf(request.getMethod().toString());
// 获得 url
final String url = requestUrl.toString();
// 获取员请求头组成httpHeaders
final DefaultHttpHeaders httpHeaders = new DefaultHttpHeaders();
request.getHeaders().forEach(httpHeaders::set);
// 开始发起请求
//创建 Netty的Request 对象
return this.httpClient.request(method, url, req -> {
final HttpClientRequest proxyRequest = req.options(NettyPipeline.SendOptions::flushOnEach)
.failOnClientError(false) //code >= 400 且clientError标志为true,才会抛出异常
.headers(httpHeaders);
// 请求类型为表单的处理
if (MediaType.APPLICATION_FORM_URLENCODED.includes(request.getHeaders().getContentType())) {
return exchange.getFormData()
.flatMap(map -> proxyRequest.sendForm(form -> {
for (Map.Entry<String, List<String>> entry: map.entrySet()) {
for (String value : entry.getValue()) {
form.attr(entry.getKey(), value);
}
}
}).then())
.then(chain.filter(exchange));
}
// 其他请求类：发送请求 Header、发送请求 Body
return proxyRequest.sendHeaders() //I shouldn't need this
.send(request.getBody()
.map(DataBuffer::asByteBuffer) // Flux<DataBuffer> => ByteBuffer
.map(Unpooled::wrappedBuffer)); // ByteBuffer => Flux<DataBuffer>
//成功发送数据时触发，res为Netty client的响应，即上面的气你执行例如
}).doOnNext(res -> {
//获取原响应
ServerHttpResponse response = exchange.getResponse();
// 将 Netty Response 的响应头和响应状态赋值给原响应
HttpHeaders headers = new HttpHeaders();
res.responseHeaders().forEach(entry -> headers.add(entry.getKey(), entry.getValue()));
response.getHeaders().putAll(headers);
response.setStatusCode(HttpStatus.valueOf(res.status().code()));
// 设置 Netty Response 到 CLIENT_RESPONSE_ATTR，最后NettyWriteResponseFilter会进行处理
exchange.getAttributes().put(CLIENT_RESPONSE_ATTR, res);
}).then(chain.filter(exchange));
}
}
NettyWriteResponseFilter
* 过滤器量的顺序是第一个，都是不处理前置，只处理过滤器后置，就是说他是这个过滤器链拦截的最后一步
* 作用是Netty 回写响应网关过滤器，与NettyRoutingFilter是匹配成对的
* 流程如下：
①调用 #then(Mono) 方法，实现 After Filter 逻辑。
②从 CLIENT_RESPONSE_ATTR 中，获得 Netty Response 。这是NettyRoutingFilter保存的
③将 Netty Response 写回给客户端
public class NettyWriteResponseFilter implements GlobalFilter, Ordered {
public static final int WRITE_RESPONSE_FILTER_ORDER = -1;
@Override
public int getOrder() {
return WRITE_RESPONSE_FILTER_ORDER;
public int getOrder() {
return WRITE_RESPONSE_FILTER_ORDER;
}
@Override
public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
return chain.filter(exchange).then(Mono.defer(() -> {
// 获得Netty的 Response
HttpClientResponse clientResponse = exchange.getAttribute(CLIENT_RESPONSE_ATTR);
if (clientResponse == null) {
return Mono.empty();
}
//获取网关原请求的响应
ServerHttpResponse response = exchange.getResponse();
// 将 Netty Response 写回给客户端。
NettyDataBufferFactory factory = (NettyDataBufferFactory) response.bufferFactory();
final Flux<NettyDataBuffer> body = clientResponse.receive() //此时就可能造成阻塞了
.retain() // ByteBufFlux => ByteBufFlux
.map(factory::wrap); // ByteBufFlux => Flux<NettyDataBuffer>
return response.writeWith(body);
}));
}
}
汇聚全球AI编程工具，助力开发者即刻编程。
更多推荐
19
0- 0
已为社区贡献2条内容
所有评论(0)