本期我們主要還是講解一下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：負責轉發我們請求的過濾器。
