源码分析Gateway请求转发
本期我们主要还是讲解一下Gateway,上一期我们讲解了一下Gateway中进行路由转发的关键角色,过滤器和断言是如何被加载的,上期链接:
https://www.cnblogs.com/guoxiaoyu/p/14735706.html
好了我们废话不多说,开始今天的Gateway请求转发流程讲解,为了在讲解源码的时候,以防止大家可能会迷糊,博主专门画了一下源码流程图,链接地址:
https://www.processon.com/view/link/60c88f64e401fd4a04b7db24
上一期我们已经知道了相关类的加载,今天直接从源码开始,大家可能不太了解webflux和reactor这种响应式编程,毕竟不是主流,我们一直用的都是spring MVC,没事,我们主要讲解流程,不做过多的讲解。
大家先看下面的代码,我们今天主要的代码入口就是这里:
1 public Mono<Void> handle(ServerWebExchange exchange) { 2 if (logger.isDebugEnabled()) { 3 ServerHttpRequest request = exchange.getRequest(); 4 logger.debug("Processing " + request.getMethodValue() + " request for [" + request.getURI() + "]"); 5 } 6 if (this.handlerMappings == null) { 7 return Mono.error(HANDLER_NOT_FOUND_EXCEPTION); 8 } 9 return Flux.fromIterable(this.handlerMappings) 10 .concatMap(mapping -> mapping.getHandler(exchange)) 11 .next() 12 .switchIfEmpty(Mono.error(HANDLER_NOT_FOUND_EXCEPTION)) 13 .flatMap(handler -> invokeHandler(exchange, handler)) 14 .flatMap(result -> handleResult(exchange, result)); 15 }
第一步,我们先来看一看几个主要的类及其方法,Flux 表示的是包含 0 到 N 个元素的异步序列,Mono 表示的是包含 0 或者 1 个元素的异步序列,记住Flux 是多个元素集合,Mono 是单个元素集合就很好理解以后的源码了,以下方法注释是博主为了大家好理解而写的,具体实际的意义还是需要大家自行Google学习了。
Mono.empty();创建一个空Mono对象;
Mono.just(**);创建一个**元素的对象;
Mono.then(**);在最后执行,相当于spring的aop后置通知一样
开始我们的第一步解析:mapping.getHandler(exchange);本方法主要做的是获取路由,我们继续看一看底层源码:
1 public Mono<Object> getHandler(ServerWebExchange exchange) { 2 return getHandlerInternal(exchange).map(handler -> { 3 if (CorsUtils.isCorsRequest(exchange.getRequest())) { 4 CorsConfiguration configA = this.globalCorsConfigSource.getCorsConfiguration(exchange); 5 CorsConfiguration configB = getCorsConfiguration(handler, exchange); 6 CorsConfiguration config = (configA != null ? configA.combine(configB) : configB); 7 if (!getCorsProcessor().process(config, exchange) || 8 CorsUtils.isPreFlightRequest(exchange.getRequest())) { 9 return REQUEST_HANDLED_HANDLER; 10 } 11 } 12 return handler; 13 }); 14 }
getHandler
1 protected Mono<?> getHandlerInternal(ServerWebExchange exchange) { 2 // don't handle requests on the management port if set 3 if (managmentPort != null && exchange.getRequest().getURI().getPort() == managmentPort.intValue()) { 4 return Mono.empty(); 5 } 6 exchange.getAttributes().put(GATEWAY_HANDLER_MAPPER_ATTR, getSimpleName()); 7 8 return lookupRoute(exchange) 9 // .log("route-predicate-handler-mapping", Level.FINER) //name this 10 .flatMap((Function<Route, Mono<?>>) r -> { 11 exchange.getAttributes().remove(GATEWAY_PREDICATE_ROUTE_ATTR); 12 if (logger.isDebugEnabled()) { 13 logger.debug("Mapping [" + getExchangeDesc(exchange) + "] to " + r); 14 } 15 16 exchange.getAttributes().put(GATEWAY_ROUTE_ATTR, r); 17 return Mono.just(webHandler); 18 }).switchIfEmpty(Mono.empty().then(Mono.fromRunnable(() -> { 19 exchange.getAttributes().remove(GATEWAY_PREDICATE_ROUTE_ATTR); 20 if (logger.isTraceEnabled()) { 21 logger.trace("No RouteDefinition found for [" + getExchangeDesc(exchange) + "]"); 22 } 23 }))); 24 }
getHandlerInternal
1 //这里返回的是单个对象 2 protected Mono<Route> lookupRoute(ServerWebExchange exchange) { 3 return this.routeLocator 4 //我们一会主要看一下这个方法 5 .getRoutes() 6 //individually filter routes so that filterWhen error delaying is not a problem 7 .concatMap(route -> Mono 8 .just(route) 9 .filterWhen(r -> { 10 // add the current route we are testing 11 exchange.getAttributes().put(GATEWAY_PREDICATE_ROUTE_ATTR, r.getId()); 12 //只返回一个符合断言的路由配置,所以整个流程先匹配断言 13 return r.getPredicate().apply(exchange); 14 }) 15 //instead of immediately stopping main flux due to error, log and swallow it 16 .doOnError(e -> logger.error("Error applying predicate for route: "+route.getId(), e)) 17 .onErrorResume(e -> Mono.empty()) 18 ) 19 // .defaultIfEmpty() put a static Route not found 20 // or .switchIfEmpty() 21 // .switchIfEmpty(Mono.<Route>empty().log("noroute")) 22 .next() 23 //TODO: error handling 24 .map(route -> { 25 if (logger.isDebugEnabled()) { 26 logger.debug("Route matched: " + route.getId()); 27 } 28 validateRoute(route, exchange); 29 return route; 30 }); 31 32 33 }
我们现在看看Route对象是怎么在getRoutes()创建的。
1 public Flux<Route> getRoutes() { 2 3 return this.routeDefinitionLocator.getRouteDefinitions() //这一步是从配置文件中读取我们配置的路由定义 4 .map(this::convertToRoute)//这一步会加载我们配置给路由的断言与过滤器形成路由对象 5 //TODO: error handling 6 .map(route -> { 7 if (logger.isDebugEnabled()) { 8 logger.debug("RouteDefinition matched: " + route.getId()); 9 } 10 return route; 11 }); 12 13 }
1 //关键的代码在这里 2 private Route convertToRoute(RouteDefinition routeDefinition) { 3 //这两步才会跟上一章节讲解的如何加载断言与过滤器有关联,大家可以自行查看底层源码是如何查出来的对象的 4 AsyncPredicate<ServerWebExchange> predicate = combinePredicates(routeDefinition); 5 List<GatewayFilter> gatewayFilters = getFilters(routeDefinition); 6 //终于生成了路由对象 7 return Route.async(routeDefinition) 8 .asyncPredicate(predicate) 9 .replaceFilters(gatewayFilters) 10 .build(); 11 }
这里大家要记住getHandlerInternal方法,生成了Mono.just(webHandler),仔细看webHandler是FilteringWebHandler对象,以后用到这个WebHandler,好了路由生成也选择完毕了,我们应该知道改请求是否符合我们配置的过滤器了,因为过滤器还没用上,断言只负责了选择哪一个路由生效。
1 //我们看下一个主流程的方法 2 private Mono<HandlerResult> invokeHandler(ServerWebExchange exchange, Object handler) { 3 if (this.handlerAdapters != null) { 4 for (HandlerAdapter handlerAdapter : this.handlerAdapters) { 5 if (handlerAdapter.supports(handler)) { 6 //这里走的是SimpleHandlerAdapter,可以自己debug发现,也可以去找自动配置类找,这里就不讲解了 7 return handlerAdapter.handle(exchange, handler); 8 } 9 } 10 } 11 return Mono.error(new IllegalStateException("No HandlerAdapter: " + handler)); 12 }
1 public Mono<HandlerResult> handle(ServerWebExchange exchange, Object handler) { 2 WebHandler webHandler = (WebHandler) handler; 3 //让大家记住的那个FilteringWebHandler类,终于在这里起作用了。我们这回可以看看过滤器是如何起作用的 4 Mono<Void> mono = webHandler.handle(exchange); 5 return mono.then(Mono.empty());//过滤器处理完后,开始处理mono.then方法 6 }
1 public Mono<Void> handle(ServerWebExchange exchange) { 2 Route route = exchange.getRequiredAttribute(GATEWAY_ROUTE_ATTR); 3 List<GatewayFilter> gatewayFilters = route.getFilters();//我们路由自己配置的过滤器 4 //加载全局过滤器 5 List<GatewayFilter> combined = new ArrayList<>(this.globalFilters); 6 combined.addAll(gatewayFilters); 7 //TODO: needed or cached? 8 AnnotationAwareOrderComparator.sort(combined); 9 //排序 10 if (logger.isDebugEnabled()) { 11 logger.debug("Sorted gatewayFilterFactories: "+ combined); 12 } 13 //形成过滤器链,开始调用filter进行过滤。这里剩下的我们就不讲解,跟spring配置的过滤器链调用流程是一样的 14 return new DefaultGatewayFilterChain(combined).filter(exchange); 15 }
至此,我们的请求流程基本完事了,我们再来看看几个主要的全局过滤器配置。LoadBalancerClientFilter:负责获取服务器ip的过滤器,NettyRoutingFilter:负责转发我们请求的过滤器。
这里主要讲解Gateway流程,关于Ribbon的代码我们就不做主要讲解了
1 public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) { 2 URI url = exchange.getAttribute(GATEWAY_REQUEST_URL_ATTR); 3 String schemePrefix = exchange.getAttribute(GATEWAY_SCHEME_PREFIX_ATTR); 4 //所以要加上lb前缀,才会走该过滤器 5 if (url == null || (!"lb".equals(url.getScheme()) && !"lb".equals(schemePrefix))) { 6 return chain.filter(exchange); 7 } 8 //preserve the original url 9 addOriginalRequestUrl(exchange, url); 10 11 log.trace("LoadBalancerClientFilter url before: " + url); 12 //选择实例 13 final ServiceInstance instance = choose(exchange); 14 15 ...... 16 return chain.filter(exchange); 17 }
看主要代码即可,非必要的看来也晕。
1 public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) { 2 3 ....... 4 //通过httpClient发送请求获取响应 5 Mono<HttpClientResponse> responseMono = this.httpClient.request(method, url, req -> { 6 final HttpClientRequest proxyRequest = req.options(NettyPipeline.SendOptions::flushOnEach) 7 .headers(httpHeaders) 8 .chunkedTransfer(chunkedTransfer) 9 .failOnServerError(false) 10 .failOnClientError(false); 11 12 if (preserveHost) { 13 String host = request.getHeaders().getFirst(HttpHeaders.HOST); 14 proxyRequest.header(HttpHeaders.HOST, host); 15 } 16 17 if (properties.getResponseTimeout() != null) { 18 proxyRequest.context(ctx -> ctx.addHandlerFirst( 19 new ReadTimeoutHandler(properties.getResponseTimeout().toMillis(), TimeUnit.MILLISECONDS))); 20 } 21 22 return proxyRequest.sendHeaders() //I shouldn't need this 23 .send(request.getBody().map(dataBuffer -> 24 ((NettyDataBuffer) dataBuffer).getNativeBuffer())); 25 }); 26 27 return responseMono.doOnNext(res -> { 28 ... 29 } 30 31 }
我们今天主要看的是Gateway的主要请求转发的流程,像webflux这种我们没有精力学习的,可以暂时略过,毕竟也不是主流。我们今天最后总结一下。首先在Gateway这两章的点,项目启动时加载断言与过滤器->接收请求时添加配置文件中的路由配置并生成路由对象->找到符合断言的路由->除了个人配置的过滤器联合全局过滤器生成过滤器链,并逐步过滤知道所有调用完成。
其中我们主要分析了两个主要的全局过滤器:LoadBalancerClientFilter:负责获取服务器ip的过滤器,NettyRoutingFilter:负责转发我们请求的过滤器。