SpringCloud gateway原理分析

秒速五厘米 2021-09-03 05:44 324阅读 0赞

cloud-gateway源码分析：

gateway-filter、global-filter：[https://blog.csdn.net/weixin\_42073629/article/details/106934370][https_blog.csdn.net_weixin_42073629_article_details_106934370]

NettyRoutingFilter、NettyWriteResponseFilter： [https://blog.csdn.net/weixin\_42073629/article/details/106934611][https_blog.csdn.net_weixin_42073629_article_details_106934611]

## 前置说明 ##

本文源码基于 `springcloud Hoxton.SR8` 版本. 以下分析仅代表个人的理解, 如有错误, 欢迎探讨.

## 使用说明 ##

以spring一贯的作风, 用户使用的模式都是约定俗成的, 也就是引入`spring-cloud-starter-gateway` 依赖, 然后就可以愉快的自动配置了. 但是这里有一个要说的点, springcloud gateway是只能使用在 webflux上面, 也就是说如果引入的springmvc依赖, 那么需要修改,不然只能使用zuul了.

<dependency>
                <groupId>org.springframework.cloud</groupId>
                <artifactId>spring-cloud-starter-gateway</artifactId>
       </dependency>

接下来就是根据自己的需求配置一些 路由的策略或者条件

spring:
      cloud:
        gateway:
          discovery:
            locator:
              enabled: false  #表明gateway开启服务注册和发现的功能，并且spring cloud gateway自动根据服务发现为每一个服务创建了一个router，这个router将以服务名开头的请求路径转发到对应的服务。
              lowerCaseServiceId: true   #是将请求路径上的服务名配置为小写（因为服务注册的时候，向注册中心注册时将服务名转成大写的了），比如以/service-hi/*的请求路径被路由转发到服务名为service-hi的服务上。
              filters:
                - StripPrefix=1
          routes:
            - id: client2
              uri: lb://client2
              predicates:
                - Path=/client2/**
              filters:
                - StripPrefix=1

## 原理介绍 ##

配置的加载

gateway的自动装配类主要在`GatewayAutoConfiguration`, 其配置了`GatewayProperties`, `FilteringWebHandler`, `RouteLocator`, `RouteDefinitionLocator`, 以及过滤器工厂和路由工厂等bean的配置

然后, 我们在配置文件中的路由配置的会被自动加载到 `GatewayProperties`, 然后在转换为`RouteDefinition` 存储在 `GatewayProperties#routes`中

下面的图是调试中一个`RouteDefinition` , 其 主要包含了 predicate 和 filters  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NkeTE5OTY_size_16_color_FFFFFF_t_70]

路由的获取

首先我们的入口在 `org.springframework.web.reactive.handler.AbstractHandlerMapping#getHandler` .

`AbstractHandlerMapping` 这个类我想应该大家看着很熟, 因为它很像`springmvc` 里面的 `handlerMapping`. 我认为这是 Spring团队想要webflux 和 springmvc 以相同的结构, 这样大家可以没有太大的学习成本来上手.

这边接下来会走到 `org.springframework.cloud.gateway.handler.RoutePredicateHandlerMapping#getHandlerInternal`, 这个类时gateway 实现的 `AbstractHandlerMapping` 来用于实现网关路由功能

@Override
    	protected Mono<?> getHandlerInternal(ServerWebExchange exchange) {
    		if (managmentPort != null && exchange.getRequest().getURI().getPort() == managmentPort.intValue()) {
    			return Mono.empty();
    		}
    		exchange.getAttributes().put(GATEWAY_HANDLER_MAPPER_ATTR, getSimpleName());
    		return lookupRoute(exchange) //超找所有的符合条件的路由
    				.flatMap((Function<Route, Mono<?>>) r -> {
    					exchange.getAttributes().remove(GATEWAY_PREDICATE_ROUTE_ATTR);
    					exchange.getAttributes().put(GATEWAY_ROUTE_ATTR, r);
    					return Mono.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) + "]");
    					}
    				})));
    	}
    
    
    
    protected Mono<Route> lookupRoute(ServerWebExchange exchange) {
    		return this.routeLocator
    				.getRoutes() // 从路由管理器中获取所有的路由
    				// 转换和连接的操作符
    				.concatMap(route -> Mono
    						.just(route)
    						.filterWhen(r -> { //过滤符合条件的route
    							exchange.getAttributes().put(GATEWAY_PREDICATE_ROUTE_ATTR, r.getId());
    							return r.getPredicate().apply(exchange); //判断路由是否符合条件
    						})
    						.doOnError(e -> logger.error("Error applying predicate for route: "+route.getId(), e))
    						.onErrorResume(e -> Mono.empty())
    				)
    				.next() //返回第一个
    				.map(route -> {
    					validateRoute(route, exchange);
    					return route;
    				});
    	}

上面看到, 主要是通过RouteLocator.getRoutes 的方法获取所有的路由, 默认的话是通过`RouteDefinitionRouteLocator` ,也就是通过`RouteDefinition` 的方式进行配置的路由. 当然我们也可以通过自定义`RouteLocator` 进行路由的提供, 然后通过`CompositeRouteLocator` 方式容器会进行自己的组装.

下面看下 `org.springframework.cloud.gateway.route.RouteDefinitionRouteLocator#getRoutes`

@Override
    	public Flux<Route> getRoutes() {
    		return this.routeDefinitionLocator.getRouteDefinitions() // 获取配置和持久化的路由
    				.map(this::convertToRoute)
    				//TODO: error handling
    				.map(route -> {
    					if (logger.isDebugEnabled()) {
    						logger.debug("RouteDefinition matched: " + route.getId());
    					}
    					return route;
    				});
    	}

路由的话前面也提到了就是可以配置在配置文件中, 也就是 `PropertiesRouteDefinitionLocator` , 当然spring也提供了其他的配置方式, 比如 `InMemoryRouteDefinitionRepository`的方法, 这个方式可以通过 REST接口进行动态的添加了路由, 也可以自定义持久化方式, 进行数据库的存储等.

public class InMemoryRouteDefinitionRepository implements RouteDefinitionRepository {
    	// 可以看到只是存储在map中
    	private final Map<String, RouteDefinition> routes = synchronizedMap(new LinkedHashMap<String, RouteDefinition>());
    
    	@Override
    	public Mono<Void> save(Mono<RouteDefinition> route) {
    		return route.flatMap( r -> {
    			routes.put(r.getId(), r);
    			return Mono.empty();
    		});
    	}
    
    	@Override
    	public Mono<Void> delete(Mono<String> routeId) {
    		return routeId.flatMap(id -> {
    			if (routes.containsKey(id)) {
    				routes.remove(id);
    				return Mono.empty();
    			}
    			return Mono.defer(() -> Mono.error(new NotFoundException("RouteDefinition not found: "+routeId)));
    		});
    	}
    
    	@Override
    	public Flux<RouteDefinition> getRouteDefinitions() {
    		return Flux.fromIterable(routes.values());
    	}
    }

路由对象的转换RouteDefinition -> Route

private Route convertToRoute(RouteDefinition routeDefinition) {
    		AsyncPredicate<ServerWebExchange> predicate = combinePredicates(routeDefinition); //  获取路由判断器
    		List<GatewayFilter> gatewayFilters = getFilters(routeDefinition); // 获取该路由要的过滤器
    
    		return Route.async(routeDefinition)
    				.asyncPredicate(predicate)
    				.replaceFilters(gatewayFilters)
    				.build();
    	}

这里的话简单说一下, 就是配置文件中的

predicates:
                - Path=/client2/**  这里的key =value , 对应了一个 RoutePredicateFactory
              filters:
                - StripPrefix=1  这里的key =value, 对应了一个 GatewayFilterFactory

按照我的配置文件写的 route包含的 predicateFactory 就是`PathRoutePredicateFactory`, filterFactory就是`StripPrefixGatewayFilterFactory`, 通过这两个工厂会返回真正的 predicate 和 filter.

流程的执行

上面的话已经讲到了请求进来根据URL来查找对应的handler, 所以接下来就是判断路由以及执行过滤器的逻辑.

Predicate的执行逻辑前面有提到,就是 `org.springframework.cloud.gateway.handler.RoutePredicateHandlerMapping#lookupRoute` 中的这一行

return r.getPredicate().apply(exchange); //判断路由是否符合条件

这里可以看下我配置的predicate

@Override
    	public Predicate<ServerWebExchange> apply(Config config) {
    		synchronized (this.pathPatternParser) {
    			pathPatternParser.setMatchOptionalTrailingSeparator(config.isMatchOptionalTrailingSeparator());
    			config.pathPattern = this.pathPatternParser.parse(config.pattern);
    		}
    		return exchange -> {
    			// 解析路径
    			PathContainer path = parsePath(exchange.getRequest().getURI().getPath());
    			//路径匹配
    			boolean match = config.pathPattern.matches(path);
    			traceMatch("Pattern", config.pathPattern.getPatternString(), path, match);
    			if (match) {
    				PathMatchInfo uriTemplateVariables = config.pathPattern.matchAndExtract(path);
    				exchange.getAttributes().put(URI_TEMPLATE_VARIABLES_ATTRIBUTE, uriTemplateVariables);
    				return true;
    			} else {
    				return false;
    			}
    		};
    	}

接下来看下 filter的执行, 也就是webHandler的执行过程  
`org.springframework.cloud.gateway.handler.FilteringWebHandler#handle`

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);
    		if (logger.isDebugEnabled()) {
    			logger.debug("Sorted gatewayFilterFactories: "+ combined);
    		}
    		// 构建过滤器链 以及执行
    		return new DefaultGatewayFilterChain(combined).filter(exchange);
    	}

可以看下我这边配置 ScriptFilter

@Override
    	public GatewayFilter apply(Config config) {
    		return (exchange, chain) ->  {
    			ServerHttpRequest request = exchange.getRequest();
    			addOriginalRequestUrl(exchange, request.getURI());
    			String path = request.getURI().getRawPath();
    			String newPath = "/" + Arrays.stream(StringUtils.tokenizeToStringArray(path, "/"))
    					.skip(config.parts).collect(Collectors.joining("/")); //这里通过skip 跳过指定的不需要的 uri段
    			ServerHttpRequest newRequest = request.mutate()
    					.path(newPath)
    					.build();
    
    			exchange.getAttributes().put(GATEWAY_REQUEST_URL_ATTR, newRequest.getURI());
    
    			return chain.filter(exchange.mutate().request(newRequest).build());
    		};
    	}

下面是进行负载均衡的过滤器`LoadBalancerClientFilter`

@Override
    	public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
    		URI url = exchange.getAttribute(GATEWAY_REQUEST_URL_ATTR);
    		String schemePrefix = exchange.getAttribute(GATEWAY_SCHEME_PREFIX_ATTR);
    		// 这里就是我配置文件中    uri: lb://client2
    		if (url == null || (!"lb".equals(url.getScheme()) && !"lb".equals(schemePrefix))) {
    			return chain.filter(exchange);
    		}
    		//preserve the original url
    		addOriginalRequestUrl(exchange, url);
    		// 根据负载均衡算法获取一个真正的远程服务器
    		final ServiceInstance instance = choose(exchange);
    
    		if (instance == null) {
    			throw new NotFoundException("Unable to find instance for " + url.getHost());
    		}
    
    		URI uri = exchange.getRequest().getURI();
    
    		String overrideScheme = null;
    		if (schemePrefix != null) {
    			overrideScheme = url.getScheme();
    		}
    		// 解析服务为具体的地址
    		URI requestUrl = loadBalancer.reconstructURI(new DelegatingServiceInstance(instance, overrideScheme), uri);
    
    		exchange.getAttributes().put(GATEWAY_REQUEST_URL_ATTR, requestUrl);
    		return chain.filter(exchange);
    	}

最后是负责 服务调用的 NettyRoutingFilter

public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
    		URI requestUrl = exchange.getRequiredAttribute(GATEWAY_REQUEST_URL_ATTR);
    
    		String scheme = requestUrl.getScheme();
    		if (isAlreadyRouted(exchange)
    				|| (!"http".equals(scheme) && !"https".equals(scheme))) {
    			return chain.filter(exchange);
    		}
    		setAlreadyRouted(exchange);
    
    		ServerHttpRequest request = exchange.getRequest();
    
    		final HttpMethod method = HttpMethod.valueOf(request.getMethodValue());
    		final String url = requestUrl.toASCIIString();
    
    		HttpHeaders filtered = filterRequest(getHeadersFilters(), exchange);
    
    		final DefaultHttpHeaders httpHeaders = new DefaultHttpHeaders();
    		filtered.forEach(httpHeaders::set);
    
    		boolean preserveHost = exchange
    				.getAttributeOrDefault(PRESERVE_HOST_HEADER_ATTRIBUTE, false);
    		Route route = exchange.getAttribute(GATEWAY_ROUTE_ATTR);
    
    		Flux<HttpClientResponse> responseFlux = getHttpClient(route, exchange)
    				.headers(headers -> {
    					headers.add(httpHeaders);
    					// Will either be set below, or later by Netty
    					headers.remove(HttpHeaders.HOST);
    					if (preserveHost) {
    						String host = request.getHeaders().getFirst(HttpHeaders.HOST);
    						headers.add(HttpHeaders.HOST, host);
    					}
    				}).request(method).uri(url).send((req, nettyOutbound) -> {
    					if (log.isTraceEnabled()) {
    						nettyOutbound
    								.withConnection(connection -> log.trace("outbound route: "
    										+ connection.channel().id().asShortText()
    										+ ", inbound: " + exchange.getLogPrefix()));
    					}
    					return nettyOutbound.send(request.getBody().map(this::getByteBuf));
    				}).responseConnection((res, connection) -> {
    
    					// Defer committing the response until all route filters have run
    					// Put client response as ServerWebExchange attribute and write
    					// response later NettyWriteResponseFilter
    					exchange.getAttributes().put(CLIENT_RESPONSE_ATTR, res);
    					exchange.getAttributes().put(CLIENT_RESPONSE_CONN_ATTR, connection);
    
    					ServerHttpResponse response = exchange.getResponse();
    					// put headers and status so filters can modify the response
    					HttpHeaders headers = new HttpHeaders();
    
    					res.responseHeaders().forEach(
    							entry -> headers.add(entry.getKey(), entry.getValue()));
    
    					String contentTypeValue = headers.getFirst(HttpHeaders.CONTENT_TYPE);
    					if (StringUtils.hasLength(contentTypeValue)) {
    						exchange.getAttributes().put(ORIGINAL_RESPONSE_CONTENT_TYPE_ATTR,
    								contentTypeValue);
    					}
    
    					setResponseStatus(res, response);
    
    					// make sure headers filters run after setting status so it is
    					// available in response
    					HttpHeaders filteredResponseHeaders = HttpHeadersFilter.filter(
    							getHeadersFilters(), headers, exchange, Type.RESPONSE);
    
    					if (!filteredResponseHeaders
    							.containsKey(HttpHeaders.TRANSFER_ENCODING)
    							&& filteredResponseHeaders
    									.containsKey(HttpHeaders.CONTENT_LENGTH)) {
    						// It is not valid to have both the transfer-encoding header and
    						// the content-length header.
    						// Remove the transfer-encoding header in the response if the
    						// content-length header is present.
    						response.getHeaders().remove(HttpHeaders.TRANSFER_ENCODING);
    					}
    
    					exchange.getAttributes().put(CLIENT_RESPONSE_HEADER_NAMES,
    							filteredResponseHeaders.keySet());
    
    					response.getHeaders().putAll(filteredResponseHeaders);
    
    					return Mono.just(res);
    				});
    
    		Duration responseTimeout = getResponseTimeout(route);
    		if (responseTimeout != null) {
    			responseFlux = responseFlux
    					.timeout(responseTimeout, Mono.error(new TimeoutException(
    							"Response took longer than timeout: " + responseTimeout)))
    					.onErrorMap(TimeoutException.class,
    							th -> new ResponseStatusException(HttpStatus.GATEWAY_TIMEOUT,
    									th.getMessage(), th));
    		}
    
    		return responseFlux.then(chain.filter(exchange));
    	}

## 总结 ##

主要的知识点

1.  gateway的配置
2.  路由的配置转换为routeDefinition
3.  获取请求对应的路由规则, 将RouteDefinition转换为Route
4.  执行Predicate判断是否符合路由, 以及执行相关的过滤(全局过滤器以及路由过滤器)
5.  负载均衡过滤器负责将请求中的serviceId转换为具体的服务实例Ip【LoadBalancerClientFilter】
6.  使用webclient filter进行实际的服务调用并获取返回结果 【NettyRoutingFilter】

[https_blog.csdn.net_weixin_42073629_article_details_106934370]: https://blog.csdn.net/weixin_42073629/article/details/106934370
[https_blog.csdn.net_weixin_42073629_article_details_106934611]: https://blog.csdn.net/weixin_42073629/article/details/106934611
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NkeTE5OTY_size_16_color_FFFFFF_t_70]: /images/20210527/896b9fed31f64d37a4b2e1574a1743a0.png