Skip to content

Gateway.md

Latest commit

History

History
475 lines (385 loc) · 18.1 KB

Gateway.md

File metadata and controls

475 lines (385 loc) · 18.1 KB
Raw

Gateway

前置

server:
port: 9002# 服务端口spring:
cloud:
gateway:
routes:
- id: user-service-impl # 路由IDuri: lb://user-service-impl # 匹配的服务名称predicates:
- Path=/user/** # 断言，路径相匹配的进行路由 -> http://user-service-impl/user/...
- id: order-service-impl # 路由IDuri: lb://order-service-impl # 匹配的服务名称predicates:
- Path=/order/** # 断言，路径相匹配的进行路由 -> http://order-service-impl/order/...
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-starter-gateway</artifactId>
</dependency>

Starters可以理解为启动器，它包含了一系列可以集成到应用里面的依赖包。所以它大概率会和SpringBoot自动装配有联系，找一下spring.factories文件。

# Auto Configure
org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
org.springframework.cloud.gateway.config.GatewayClassPathWarningAutoConfiguration,\
org.springframework.cloud.gateway.config.GatewayAutoConfiguration,\
org.springframework.cloud.gateway.config.GatewayResilience4JCircuitBreakerAutoConfiguration,\
org.springframework.cloud.gateway.config.GatewayNoLoadBalancerClientAutoConfiguration,\
org.springframework.cloud.gateway.config.GatewayMetricsAutoConfiguration,\
org.springframework.cloud.gateway.config.GatewayRedisAutoConfiguration,\
org.springframework.cloud.gateway.discovery.GatewayDiscoveryClientAutoConfiguration,\
org.springframework.cloud.gateway.config.SimpleUrlHandlerMappingGlobalCorsAutoConfiguration,\
org.springframework.cloud.gateway.config.GatewayReactiveLoadBalancerClientAutoConfiguration,\
org.springframework.cloud.gateway.config.GatewayReactiveOAuth2AutoConfiguration
org.springframework.boot.env.EnvironmentPostProcessor=\
org.springframework.cloud.gateway.config.GatewayEnvironmentPostProcessor
# Failure Analyzers
org.springframework.boot.diagnostics.FailureAnalyzer=\
org.springframework.cloud.gateway.support.MvcFoundOnClasspathFailureAnalyzer

packageorg.springframework.cloud.gateway.config;
@Configuration(proxyBeanMethods = false)
// 自动装配生效@ConditionalOnProperty(name = "spring.cloud.gateway.enabled", matchIfMissing = true)
@EnableConfigurationProperties// 自动配置前置条件@AutoConfigureBefore({ HttpHandlerAutoConfiguration.class, WebFluxAutoConfiguration.class })
// 自动配置后置组件@AutoConfigureAfter({ GatewayReactiveLoadBalancerClientAutoConfiguration.class,
GatewayClassPathWarningAutoConfiguration.class })
@ConditionalOnClass(DispatcherHandler.class)
publicclassGatewayAutoConfiguration {
// ...会注入很多 Bean
}

从配置类上定义的注解可以了解到：

  • spring.cloud.gateway.enabled配置项为true时，自动配置才会生效，默认true。

  • 在注入Gateway之前，容器中必须存在 WebFlux 和 HttpHandler 组件。

  • 在注入Gateway之后，需要对请求负载均衡，所以后置组件为负载均衡组件。

GatewayAutoConfiguration在内部会初始化很多bean，这里列举几个重要的：

  • PropertiesRouteDefinitionLocator：用于从配置文件（yaml/properties）中读取路由配置信息。

  • RouteDefinitionLocator：把 RouteDefinition 转化为 Route。

  • RoutePredicateHandlerMapping：类似于 MVC 的HandlerMapping，不过这里是 Gateway 实现的。用于匹配对应的请求route。

  • GatewayProperties：yaml配置信息被封装在 GatewayProperties 对象中。

  • AfterRoutePredicateFactory：路由断言工厂。

  • RetryGatewayFilterFactory：重试 Gateway 过滤器工厂。

  • GlobalFilter：全局过滤器。

Gateway 的执行流程大致如下图所示：

其中GatewayAutoConfiguration是Gateway的自动配置类。

Gateway调用过程

Gateway采用的是 webFlux 响应式编程，其请求处理流程与Spring MVC类似。

GatewaySpring MVC
请求分发DispatcherHandlerDispatcherServlet
请求映射HandlerMappingHandlerMapping
请求适配HandlerAdapterHandlerAdapter
请求处理WebHandlerHandler

所有的请求都会经过 Gateway 的DispatcherHandler#handler()方法。

publicclassDispatcherHandlerimplementsWebHandler, ApplicationContextAware {
@OverridepublicMono<Void> handle(ServerWebExchangeexchange) {
if (this.handlerMappings == null) {
returncreateNotFoundError();
}
returnFlux// 遍历所有的 handlerMapping
.fromIterable(this.handlerMappings)
// 获取到请求映射的 handler，最终获取到的是 FilteringWebHandler
.concatMap(mapping -> mapping.getHandler(exchange))
.next()
.switchIfEmpty(createNotFoundError())
// 获取对应的适配器处理器，执行目标处理器方法
.flatMap(handler -> invokeHandler(exchange, handler))
// 返回处理结果
.flatMap(result -> handleResult(exchange, result));
}
}

路由断言

AbstractHandlerMapping.getHandler()

publicabstractclassAbstractHandlerMappingextendsApplicationObjectSupportimplementsHandlerMapping, Ordered, BeanNameAware {
@OverridepublicMono<Object> getHandler(ServerWebExchangeexchange) {
// 获取匹配的路由 RoutereturngetHandlerInternal(exchange).map(handler -> {
if (logger.isDebugEnabled()) {
logger.debug(exchange.getLogPrefix() + "Mapped to " + handler);
}
ServerHttpRequestrequest = exchange.getRequest();
if (hasCorsConfigurationSource(handler) || CorsUtils.isPreFlightRequest(request)) {
CorsConfigurationconfig = (this.corsConfigurationSource != null ? this.corsConfigurationSource.getCorsConfiguration(exchange) : null);
CorsConfigurationhandlerConfig = getCorsConfiguration(handler, exchange);
config = (config != null ? config.combine(handlerConfig) : handlerConfig);
if (!this.corsProcessor.process(config, exchange) || CorsUtils.isPreFlightRequest(request)) {
returnREQUEST_HANDLED_HANDLER;
}
}
returnhandler;
});
}
}

由于是网关组件，当请求进入时，会先判断路由，所以会进入实现类 RoutePredicateHandlerMapping 的逻辑中。

publicclassRoutePredicateHandlerMappingextendsAbstractHandlerMapping {
@OverrideprotectedMono<?> getHandlerInternal(ServerWebExchangeexchange) {
// don't handle requests on management port if set and different than server portif (this.managementPortType == DIFFERENT && this.managementPort != null
&& exchange.getRequest().getURI().getPort() == this.managementPort) {
returnMono.empty();
}
exchange.getAttributes().put(GATEWAY_HANDLER_MAPPER_ATTR, getSimpleName());
// 查找路由，包含自定义FilterreturnlookupRoute(exchange)
// .log("route-predicate-handler-mapping", Level.FINER) //name this
.flatMap((Function<Route, Mono<?>>) r -> {
// 除上下文中旧的属性exchange.getAttributes().remove(GATEWAY_PREDICATE_ROUTE_ATTR);
if (logger.isDebugEnabled()) {
logger.debug("Mapping [" + getExchangeDesc(exchange) + "] to " + r);
}
// 将匹配到的路由与上下文绑定，用于后续的负载均衡exchange.getAttributes().put(GATEWAY_ROUTE_ATTR, r);
returnMono.just(webHandler);
}).switchIfEmpty(Mono.empty().then(Mono.fromRunnable(() -> {
exchange.getAttributes().remove(GATEWAY_PREDICATE_ROUTE_ATTR);
if (logger.isTraceEnabled()) {
logger.trace("No RouteDefinition found for [" + getExchangeDesc(exchange) + "]");
}
})));
}
}

**lookupRoute()**会找到配置的所有路由断言工厂，然后判断是否允许请求通过。

publicclassRoutePredicateHandlerMappingextendsAbstractHandlerMapping {
protectedMono<Route> lookupRoute(ServerWebExchangeexchange) {
// getRoutes > 获取配置的所有断言工厂returnthis.routeLocator.getRoutes()
// individually filter routes so that filterWhen error delaying is not a// problem
.concatMap(route -> Mono.just(route).filterWhen(r -> {
// add the current route we are testingexchange.getAttributes().put(GATEWAY_PREDICATE_ROUTE_ATTR, r.getId());
// 先获取Route的predicate（断言）属性，然后调用apply()执行断言，判断请求是否允许通过returnr.getPredicate().apply(exchange);
})
// instead of immediately stopping main flux due to error, log and// swallow it
.doOnError(e -> logger.error("Error applying predicate for route: " + route.getId(), e))
.onErrorResume(e -> Mono.empty()))
// .defaultIfEmpty() put a static Route not found// or .switchIfEmpty()// .switchIfEmpty(Mono.<Route>empty().log("noroute"))
.next()
// TODO: error handling
.map(route -> {
if (logger.isDebugEnabled()) {
logger.debug("Route matched: " + route.getId());
}
validateRoute(route, exchange);
returnroute;
});
}
}

**getRoutes()**获取配置的所有路由。

publicclassRouteDefinitionRouteLocatorimplementsRouteLocator {
@OverridepublicFlux<Route> getRoutes() {
// 从配置文件中获取所有路由，并转换成 RouteFlux<Route> routes = this.routeDefinitionLocator.getRouteDefinitions().map(this::convertToRoute);
if (!gatewayProperties.isFailOnRouteDefinitionError()) {
// instead of letting error bubble up, continueroutes = routes.onErrorContinue((error, obj) -> {
if (logger.isWarnEnabled()) {
logger.warn("RouteDefinition id " + ((RouteDefinition) obj).getId()
+ " will be ignored. Definition has invalid configs, " + error.getMessage());
}
});
}
returnroutes.map(route -> {
if (logger.isDebugEnabled()) {
logger.debug("RouteDefinition matched: " + route.getId());
}
returnroute;
});
}
}
// Route内部结构publicclassRouteimplementsOrdered {
// 路由idprivatefinalStringid;
// 请求URIprivatefinalURIuri;
// 排序privatefinalintorder;
// 断言privatefinalAsyncPredicate<ServerWebExchange> predicate;
// 过滤器privatefinalList<GatewayFilter> gatewayFilters;
// 元数据privatefinalMap<String, Object> metadata;
}

适配器处理器

在前面已经获取到了 handlerMapping，处理器映射器，并执行断言。假设请求符合断言，第二步就是执行过滤器链了。

publicclassDispatcherHandlerimplementsWebHandler, ApplicationContextAware {
privateMono<HandlerResult> invokeHandler(ServerWebExchangeexchange, Objecthandler) {
if (this.handlerAdapters != null) {
// 找到所有的 handlerAdapter，匹配 WebFlux，默认会由 SimpleHandlerAdapter 处理for (HandlerAdapterhandlerAdapter : this.handlerAdapters) {
if (handlerAdapter.supports(handler)) {
returnhandlerAdapter.handle(exchange, handler);
}
}
}
returnMono.error(newIllegalStateException("No HandlerAdapter: " + handler));
}
}

SimpleHandlerAdapter.handle()

publicclassSimpleHandlerAdapterimplementsHandlerAdapter {
@OverridepublicMono<HandlerResult> handle(ServerWebExchangeexchange, Objecthandler) {
WebHandlerwebHandler = (WebHandler) handler;
Mono<Void> mono = webHandler.handle(exchange);
returnmono.then(Mono.empty());
}
}

**FilteringWebHandler.handle()**执行过滤器链，该过滤器链包括 globalFilters 和 gatewayFilters。

publicclassFilteringWebHandlerimplementsWebHandler {
@OverridepublicMono<Void> handle(ServerWebExchangeexchange) {
// 根据路由与上下文的绑定关系，获取对应的路由 RouteRouteroute = exchange.getRequiredAttribute(GATEWAY_ROUTE_ATTR);
List<GatewayFilter> gatewayFilters = route.getFilters();
List<GatewayFilter> combined = newArrayList<>(this.globalFilters);
// 全局过滤器 + gatewayFilters 形成一条过滤器链，先执行的是GlobalFilter，然后是gatewayFiltercombined.addAll(gatewayFilters);
// TODO: needed or cached?// 根据order排序，order值越小越先执行，order相同的则按照文件名排序AnnotationAwareOrderComparator.sort(combined);
if (logger.isDebugEnabled()) {
logger.debug("Sorted gatewayFilterFactories: " + combined);
}
// 构造过滤器链，执行过滤器链中的每一个过滤器方法returnnewDefaultGatewayFilterChain(combined).filter(exchange);
}
privatestaticclassGatewayFilterAdapterimplementsGatewayFilter {
// 封装了 GlobalFilterprivatefinalGlobalFilterdelegate;
GatewayFilterAdapter(GlobalFilterdelegate) {
this.delegate = delegate;
}
@OverridepublicMono<Void> filter(ServerWebExchangeexchange, GatewayFilterChainchain) {
returnthis.delegate.filter(exchange, chain);
}
@OverridepublicStringtoString() {
finalStringBuildersb = newStringBuilder("GatewayFilterAdapter{");
sb.append("delegate=").append(delegate);
sb.append('}');
returnsb.toString();
}
}
}

全局过滤器会被封装为 GatewayFilterAdapter，其实现了 GatewayFilter，解决了 globalFilters 想要放入List<GatewayFilter>中的类型不一致问题。

然后在执行过滤器链中的 globalFilters 和 gatewayFilters 的 filter() 方法时，就会为请求加上请求头、请求参数等扩展点。

Gateway的负载均衡是如何实现的

Gateway的负载均衡只需要在yaml中配置uri: lb://mall-order即可实现负载均衡，底层是由全局过滤器**ReactiveLoadBalancerClientFilter.filter()**方法去做的。

使用负载均衡客户端选取后端服务实例, 构建目标URL并存储在ServerWebExchange中。

publicclassReactiveLoadBalancerClientFilterimplementsGlobalFilter, Ordered {
@OverridepublicMono<Void> filter(ServerWebExchangeexchange, GatewayFilterChainchain) {
URIurl = exchange.getAttribute(GATEWAY_REQUEST_URL_ATTR);
StringschemePrefix = exchange.getAttribute(GATEWAY_SCHEME_PREFIX_ATTR);
if (url == null || (!"lb".equals(url.getScheme()) && !"lb".equals(schemePrefix))) {
returnchain.filter(exchange);
}
// preserve the original urladdOriginalRequestUrl(exchange, url);
if (log.isTraceEnabled()) {
log.trace(ReactiveLoadBalancerClientFilter.class.getSimpleName() + " url before: " + url);
}
URIrequestUri = exchange.getAttribute(GATEWAY_REQUEST_URL_ATTR);
StringserviceId = requestUri.getHost();
Set<LoadBalancerLifecycle> supportedLifecycleProcessors = LoadBalancerLifecycleValidator
.getSupportedLifecycleProcessors(clientFactory.getInstances(serviceId, LoadBalancerLifecycle.class),
RequestDataContext.class, ResponseData.class, ServiceInstance.class);
DefaultRequest<RequestDataContext> lbRequest = newDefaultRequest<>(
newRequestDataContext(newRequestData(exchange.getRequest()), getHint(serviceId)));
LoadBalancerPropertiesloadBalancerProperties = clientFactory.getProperties(serviceId);
// 使用 doOnNext() 处理 choose 方法返回的 ServiceInstance 实例// choose() 选择服务的逻辑和 OpenFeign 类似，如果使用的是 Nacos，则调用 Nacos 的组件returnchoose(lbRequest, serviceId, supportedLifecycleProcessors).doOnNext(response -> {
if (!response.hasServer()) {
supportedLifecycleProcessors.forEach(lifecycle -> lifecycle
.onComplete(newCompletionContext<>(CompletionContext.Status.DISCARD, lbRequest, response)));
throwNotFoundException.create(properties.isUse404(), "Unable to find instance for " + url.getHost());
}
ServiceInstanceretrievedInstance = response.getServer();
URIuri = exchange.getRequest().getURI();
// if the `lb:<scheme>` mechanism was used, use `<scheme>` as the default,// if the loadbalancer doesn't provide one.// 协议StringoverrideScheme = retrievedInstance.isSecure() ? "https" : "http";
if (schemePrefix != null) {
overrideScheme = url.getScheme();
}
DelegatingServiceInstanceserviceInstance = newDelegatingServiceInstance(retrievedInstance,
overrideScheme);
// 构建目标请求 uri 将服务名替换为 ipURIrequestUrl = reconstructURI(serviceInstance, uri);
if (log.isTraceEnabled()) {
log.trace("LoadBalancerClientFilter url chosen: " + requestUrl);
}
// 将构建的目标请求URI存储到ServerWebExchange的属性中提供给后续的过滤器exchange.getAttributes().put(GATEWAY_REQUEST_URL_ATTR, requestUrl);
exchange.getAttributes().put(GATEWAY_LOADBALANCER_RESPONSE_ATTR, response);
supportedLifecycleProcessors.forEach(lifecycle -> lifecycle.onStartRequest(lbRequest, response));
}).then(chain.filter(exchange))
.doOnError(throwable -> supportedLifecycleProcessors.forEach(lifecycle -> lifecycle
.onComplete(newCompletionContext<ResponseData, ServiceInstance, RequestDataContext>(
CompletionContext.Status.FAILED, throwable, lbRequest,
exchange.getAttribute(GATEWAY_LOADBALANCER_RESPONSE_ATTR)))))
.doOnSuccess(aVoid -> supportedLifecycleProcessors.forEach(lifecycle -> lifecycle
.onComplete(newCompletionContext<ResponseData, ServiceInstance, RequestDataContext>(
CompletionContext.Status.SUCCESS, lbRequest,
exchange.getAttribute(GATEWAY_LOADBALANCER_RESPONSE_ATTR), buildResponseData(exchange,
loadBalancerProperties.isUseRawStatusCodeInResponseData())))));
}
privateMono<Response<ServiceInstance>> choose(Request<RequestDataContext> lbRequest, StringserviceId,
Set<LoadBalancerLifecycle> supportedLifecycleProcessors) {
// 获取负载策略ReactorLoadBalancer<ServiceInstance> loadBalancer = this.clientFactory.getInstance(serviceId,
ReactorServiceInstanceLoadBalancer.class);
if (loadBalancer == null) {
thrownewNotFoundException("No loadbalancer available for " + serviceId);
}
supportedLifecycleProcessors.forEach(lifecycle -> lifecycle.onStart(lbRequest));
// NacosLoadBalancerreturnloadBalancer.choose(lbRequest);
}
}

GateWay的源码体量相对较小，但想要用一篇文章来深入也不太现实，都是挑执行链路上的核心点进行剖析。