Java源码解析之Gateway请求转发

public Mono<Void> handle(ServerWebExchange exchange) {
        if (logger.isDebugEnabled()) {
            ServerHttpRequest request = exchange.getRequest();
            logger.debug("Processing " + request.getMethodValue() + " request for [" + request.getURI() + "]");
        }
        if (this.handlerMappings == null) {
            return Mono.error(HANDLER_NOT_FOUND_EXCEPTION);
        }
        return Flux.fromIterable(this.handlerMappings)
                .concatMap(mapping -> mapping.getHandler(exchange))
                .next()
                .switchIfEmpty(Mono.error(HANDLER_NOT_FOUND_EXCEPTION))
                .flatMap(handler -> invokeHandler(exchange, handler))
                .flatMap(result -> handleResult(exchange, result));
    }

//这里返回的是单个对象
    protected Mono<Route> lookupRoute(ServerWebExchange exchange) {
        return this.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 testing
                            exchange.getAttributes().put(GATEWAY_PREDICATE_ROUTE_ATTR, r.getId());
                            //只返回一个符合断言的路由配置，所以整个流程先匹配断言
                            return r.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);
                    return route;
                });


    }

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;
                });

    }

//关键的代码在这里
    private Route convertToRoute(RouteDefinition routeDefinition) {
    //这两步才会跟上一章节讲解的如何加载断言与过滤器有关联，大家可以自行查看底层源码是如何查出来的对象的
        AsyncPredicate<ServerWebExchange> predicate = combinePredicates(routeDefinition);
        List<GatewayFilter> gatewayFilters = getFilters(routeDefinition);
    //终于生成了路由对象
        return Route.async(routeDefinition)
                .asyncPredicate(predicate)
                .replaceFilters(gatewayFilters)
                .build();
    }

//我们看下一个主流程的方法
    private Mono<HandlerResult> invokeHandler(ServerWebExchange exchange, Object handler) {
        if (this.handlerAdapters != null) {
            for (HandlerAdapter handlerAdapter : this.handlerAdapters) {
                if (handlerAdapter.supports(handler)) {
                //这里走的是SimpleHandlerAdapter，可以自己debug发现，也可以去找自动配置类找，这里就不讲解了
                    return handlerAdapter.handle(exchange, handler);
                }
            }
        }
        return Mono.error(new IllegalStateException("No HandlerAdapter: " + handler));
    }

public Mono<HandlerResult> handle(ServerWebExchange exchange, Object handler) {
        WebHandler webHandler = (WebHandler) handler;
        //让大家记住的那个FilteringWebHandler类，终于在这里起作用了。我们这回可以看看过滤器是如何起作用的
        Mono<Void> mono = webHandler.handle(exchange);
         return mono.then(Mono.empty());//过滤器处理完后，开始处理mono.then方法
    }

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);
        //TODO: needed or cached?
        AnnotationAwareOrderComparator.sort(combined);
        //排序
        if (logger.isDebugEnabled()) {
            logger.debug("Sorted gatewayFilterFactories: "+ combined);
        }
        //形成过滤器链，开始调用filter进行过滤。这里剩下的我们就不讲解，跟spring配置的过滤器链调用流程是一样的
        return new DefaultGatewayFilterChain(combined).filter(exchange);
    }

public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
        URI url = exchange.getAttribute(GATEWAY_REQUEST_URL_ATTR);
        String schemePrefix = exchange.getAttribute(GATEWAY_SCHEME_PREFIX_ATTR);
        //所以要加上lb前缀，才会走该过滤器
        if (url == null || (!"lb".equals(url.getScheme()) && !"lb".equals(schemePrefix))) {
            return chain.filter(exchange);
        }
        //preserve the original url
        addOriginalRequestUrl(exchange, url);

        log.trace("LoadBalancerClientFilter url before: " + url);
        //选择实例
        final ServiceInstance instance = choose(exchange);

        ......
        return chain.filter(exchange);
    }

public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {

        .......
        //通过httpClient发送请求获取响应
        Mono<HttpClientResponse> responseMono = this.httpClient.request(method, url, req -> {
            final HttpClientRequest proxyRequest = req.options(NettyPipeline.SendOptions::flushOnEach)
                    .headers(httpHeaders)
                    .chunkedTransfer(chunkedTransfer)
                    .failOnServerError(false)
                    .failOnClientError(false);

            if (preserveHost) {
                String host = request.getHeaders().getFirst(HttpHeaders.HOST);
                proxyRequest.header(HttpHeaders.HOST, host);
            }

            if (properties.getResponseTimeout() != null) {
                proxyRequest.context(ctx -> ctx.addHandlerFirst(
                        new ReadTimeoutHandler(properties.getResponseTimeout().toMillis(), TimeUnit.MILLISECONDS)));
            }

            return proxyRequest.sendHeaders() //I shouldn't need this
                    .send(request.getBody().map(dataBuffer ->
                            ((NettyDataBuffer) dataBuffer).getNativeBuffer()));
        });

        return responseMono.doOnNext(res -> {
        ...
        }

    }