vlambda博客
学习文章列表

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

public Mono<Object> getHandler(ServerWebExchange exchange) { return getHandlerInternal(exchange).map(handler -> { if (CorsUtils.isCorsRequest(exchange.getRequest())) { CorsConfiguration configA = this.globalCorsConfigSource.getCorsConfiguration(exchange); CorsConfiguration configB = getCorsConfiguration(handler, exchange); CorsConfiguration config = (configA != null ? configA.combine(configB) : configB); if (!getCorsProcessor().process(config, exchange) || CorsUtils.isPreFlightRequest(exchange.getRequest())) { return REQUEST_HANDLED_HANDLER; } } return handler; }); }

protected Mono<?> getHandlerInternal(ServerWebExchange exchange) { // don't handle requests on the management port if set if (managmentPort != null && exchange.getRequest().getURI().getPort() == managmentPort.intValue()) { return Mono.empty(); } exchange.getAttributes().put(GATEWAY_HANDLER_MAPPER_ATTR, getSimpleName());
 return lookupRoute(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); 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() //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);         //过滤器处理完后，开始处理mono.then方法         return mono.then(Mono.empty());    }

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 -> { ... }
 }