1、概述

本周又回顾了一下Spring Cloud Gateway的内容,主要参考《重新定义Spring Cloud实战》以及一些网络博客,虽然内容都差不多,但我还想照猫画虎总结一下子,加深印象。

SpringCloud gateway是Spring Cloud生态体系的第二代网关,是基于Spring5.0、Spring Boot 2.0、Reactor等技术开发的网关。底层主要是两大核心:Spring web Filter chain和Spring WebFlux。

对两大核心不了解的可以先补补知识,第二部分分析一下gateway中如何使用的filter,以及chain的实现:
Filter :Filter是一个Servlet规范组件;一个请求可以在Http请求到达Servlet前被一个或多个Filter处理,Servlet处理完后返回给Filter,最后返回给用户。

WebFlux:它是一个异步非阻塞式的web框架,它的作用不是提升接口请求时间,而是在一些阻塞的场景【例如请求DB,等待DB响应数据、打开大文件等】,可以把线程给其它请求使用,从而提升系统吞吐量。Gateway属于网络IO密集型【网关转发请求到下游服务】,通过WebFlux有效的提升网关转发的吞吐量。

Spring Cloud Gateway核心概念:

  • 路由(Route.java):由一个id,一个目标uri,一组断言工厂和一组Filter组成
public class Route implements Ordered {

	private final String id;

	private final URI uri;

	private final int order;

	private final AsyncPredicate<ServerWebExchange> predicate;

	private final List<GatewayFilter> gatewayFilters;
	/**省略构造函数和get set方法/
}

  • 断言(AsyncPredicate):java8中的断言函数,上述代码中的predicate属性
  • 过滤器:标准的Spring web Filter;SpringCloud Gateway中包含GlobalFilter和GatewayFilter。

2、ServerWebExchange

SpringCloud gateway的上下文是ServerWebExchange,请求的信息都存储在ServerWebExchange中,在网关上的后续操作都是基于上下文操作的,在http请求到达网关之后,网关入口是ReactorHttpHandlerAdapter#apply方法,去获取请求的request和response,构建当次请求的上下文供后续filter使用:

public class ReactorHttpHandlerAdapter implements BiFunction<HttpServerRequest, HttpServerResponse, Mono<Void>> {
   
     
	@Override
	public Mono<Void> apply(HttpServerRequest reactorRequest, HttpServerResponse reactorResponse) {
   
     
		NettyDataBufferFactory bufferFactory = new NettyDataBufferFactory(reactorResponse.alloc());
		try {
   
     
		    //获取请求的Request,构建ReactorServerHttpRequest
			ReactorServerHttpRequest request = new ReactorServerHttpRequest(reactorRequest, bufferFactory);
			//构建ServerHttpResponse
			ServerHttpResponse response = new ReactorServerHttpResponse(reactorResponse, bufferFactory);

			if (request.getMethod() == HttpMethod.HEAD) {
   
     
				response = new HttpHeadResponseDecorator(response);
			}
			//交给HttpWebHandlerAdapter构建上下文ServerWebExchange
			return this.httpHandler.handle(request, response)
					.doOnError(ex -> logger.trace(request.getLogPrefix() + "Failed to complete: " + ex.getMessage()))
					.doOnSuccess(aVoid -> logger.trace(request.getLogPrefix() + "Handling completed"));
		}
		catch (URISyntaxException ex) {
   
     
			if (logger.isDebugEnabled()) {
   
     
				logger.debug("Failed to get request URI: " + ex.getMessage());
			}
			reactorResponse.status(HttpResponseStatus.BAD_REQUEST);
			return Mono.empty();
		}
	}
}

构建完request和response后,交给HttpWebHandlerAdapter构建上下文ServerWebExchange

public class HttpWebHandlerAdapter extends WebHandlerDecorator implements HttpHandler {
   
     
	public Mono<Void> handle(ServerHttpRequest request, ServerHttpResponse response) {
   
     

		if (this.forwardedHeaderTransformer != null) {
   
     
			request = this.forwardedHeaderTransformer.apply(request);
		}
		//构建请求的上下文
		ServerWebExchange exchange = createExchange(request, response);

		LogFormatUtils.traceDebug(logger, traceOn ->
				exchange.getLogPrefix() + formatRequest(exchange.getRequest()) +
						(traceOn ? ", headers=" + formatHeaders(exchange.getRequest().getHeaders()) : ""));

		return getDelegate().handle(exchange)
				.doOnSuccess(aVoid -> logResponse(exchange))
				.onErrorResume(ex -> handleUnresolvedError(exchange, ex))
				.then(Mono.defer(response::setComplete));
	}
}

3、Route和RouteDefinition

我们在配置文件中配置的一个路由规则,对应到Java类就是GatewayProperties,Spring boot会将配置文件映射为Java类,例如如下配置:

spring:
  cloud:
    gateway:
      routes:
      - id: add_request_header_route
        uri: http://localhost:8080
        predicates:
        - Path=/test/**
        filters:
        - AddRequestHeader=NAME, test

RouteDefinition:路由定义,是GatewayProperties类中的一个属性,网关启动后,Springboot帮我们做了映射,上述配置的路由就设置到了 GatewayProperties对象中。
Route:是从路由定义

路由信息映射到GatewayProperties后如何获取其中的RouteDefinition?
答案是通过RouteDefinitionLocator,RouteDefinitionLocator有5个实现类
 

  • PropertiesRouteDefinitionLocator:从Properties中读取
public class PropertiesRouteDefinitionLocator implements RouteDefinitionLocator {
   
     

	private final GatewayProperties properties;
	//构造函数设置properties
	public PropertiesRouteDefinitionLocator(GatewayProperties properties) {
   
     
		this.properties = properties;
	}
	//从properties中读取RouteDefinition
	@Override
	public Flux<RouteDefinition> getRouteDefinitions() {
   
     
		return Flux.fromIterable(this.properties.getRoutes());
	}
}

  • InMemoryRouteDefinitionRepository:对RouteDefinition进行增、删、查操作,基于内存存储
  • CompositeRouteDefinitionLocator:组合的Locator,在构造函数中设置委托,将PropertiesRouteDefinitionLocator和InMemoryRouteDefinitionRepository组合。
public class CompositeRouteDefinitionLocator implements RouteDefinitionLocator {
   
     

	private final Flux<RouteDefinitionLocator> delegates;
	//将PropertiesRouteDefinitionLocator和InMemoryRouteDefinitionRepository组合
	public CompositeRouteDefinitionLocator(Flux<RouteDefinitionLocator> delegates) {
   
     
		this.delegates = delegates;
	}
	//委托给PropertiesRouteDefinitionLocator或InMemoryRouteDefinitionRepository执行读取
	@Override
	public Flux<RouteDefinition> getRouteDefinitions() {
   
     
		return this.delegates.flatMap(RouteDefinitionLocator::getRouteDefinitions);
	}

}

4、GlobalFilter和GatewayFilter

两种Filter都是拦截到http请求后做一些处理,gateway为什么提供两种Filter?
总结:GlobalFilter是所有被gateway拦截的http请求都要做的处理;GatewayFilter是根据路由配置匹配predicate的http请求才会做的处理。

4.1、GlobalFilter

GlobalFilter:全局拦截器,是所有被拦截到的http请求都要去做的处理;例如拿到一个http请求后,我们的目的是转发到下游服务,请求结果并返回,那么所有被拦截到的http请求都需要做下列几件事:

  • 按照predicate把符合规则的url转换为真正要去请求的url
  • 调用真正的下游服务【SpringCloud Gateway是基于netty实现的http调用,具体代码在NettyRoutingFilter类中】
  • 拿到response,返回给调用方

像这种每个被拦截到的http请求都要去做的处理抽象出来就是一个个的GlobalFilter
 

接口定义

public interface GlobalFilter {
   
     
	Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain);
}

接口中只有一个filter方法,实现类实现该接口后在filter中去做具体拦截逻辑,这些Filter都实现了GlobalFilter接口,一起来看一下实现该接口的类都做了什么操作:

  • AdaptCachedBodyGlobalFilter:优先级最高的Filter,请求到gateway后,将上下文【ServerWebExchange】中已有的缓存删除【请求信息】,将此次的请求信息缓存到上下文中。
public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
     
		// 获取上下文中已有的请求缓存
		Flux<DataBuffer> body = exchange.getAttributeOrDefault(CACHED_REQUEST_BODY_KEY,
				null);
		if (body != null) {
   
			ServerHttpRequestDecorator decorator = new ServerHttpRequestDecorator(
					exchange.getRequest()) {
     
				@Override
				public Flux<DataBuffer> getBody() {
     
					return body;
				}
			};
			// 删除上下文中的请求缓存
			exchange.getAttributes().remove(CACHED_REQUEST_BODY_KEY);
			// 将此次请求信息添加到上下文缓存中
			return chain.filter(exchange.mutate().request(decorator).build());
		}

		return chain.filter(exchange);
	}

  • GatewayMetricsFilter:统计网关的性能指标
private void endTimerInner(ServerWebExchange exchange, Sample sample) {
     
    // 此处省略一部分代码
    // ...
    Route route = exchange.getAttribute(GATEWAY_ROUTE_ATTR);
    // 统计网关的性能指标
    Tags tags = Tags.of("outcome", outcome, "status", status, "routeId",
            route.getId(), "routeUri", route.getUri().toString());
    if (log.isTraceEnabled()) {

 
        log.trace("Stopping timer 'gateway.requests' with tags " + tags);
    }
    sample.stop(meterRegistry.timer("gateway.requests", tags));
}

  • NettyWriteResponseFilter:基于Web Flux,若上下文中存在CLIENT_RESPONSE_CONN_ATTR,将响应数据返回。
  • ForwardPathFilter:如果该请求还未被路由或scheme【URI对象的属性】不是forward,则将该请求对应配置的Route信息中uri的path设置到上下文ServerWebExchange中。
public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
 
    Route route = exchange.getAttribute(GATEWAY_ROUTE_ATTR);
    URI routeUri = route.getUri();
    String scheme = routeUri.getScheme();
    if (isAlreadyRouted(exchange) || !"forward".equals(scheme)) {

        return chain.filter(exchange);
    }
    // 将该请求对应配置的Route信息中uri的path设置到上下文ServerWebExchange中
    exchange = exchange.mutate()
            .request(exchange.getRequest().mutate().path(routeUri.getPath()).build())
            .build();
    return chain.filter(exchange);
}

  • RouteToRequestUrlFilter:将此次请求的uri和配置的Route规则做merged处理,拿到真正代理的下游服务的地址,将得到的url放到上下文中,key为GATEWAY_REQUEST_URL_ATTR
  • NoLoadBalancerClientFilter:没有负载均衡的拦截器
  • WebsocketRoutingFilter:路由WebSocket请求,校验逻辑在WebsocketRoutingFilter#changeSchemeIfIsWebSocketUpgrade中。
  • NettyRoutingFilter:网关的http是基于netty实现的,若此次请求scheme是http或https则使用基于netty的httpClient执行调用,将返回结果写入上下文中。
exchange.getAttributes().put(CLIENT_RESPONSE_ATTR, res);
exchange.getAttributes().put(CLIENT_RESPONSE_CONN_ATTR, connection);

  • ForwardRoutingFilter:设置此次请求已被路由
public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
    URI requestUrl = exchange.getRequiredAttribute(GATEWAY_REQUEST_URL_ATTR);

    String scheme = requestUrl.getScheme();
    if (isAlreadyRouted(exchange) || !"forward".equals(scheme)) {
 
        return chain.filter(exchange);
    }
    // 设置此次请求已被路由
    setAlreadyRouted(exchange);

    if (log.isTraceEnabled()) {

 
        log.trace("Forwarding to URI: " + requestUrl);
    }

    return this.getDispatcherHandler().handle(exchange);
}

  • WebClientHttpRoutingFilter:作用同NettyRoutingFilter,方式同LoadBalancerClientFilter
  • WebClientWriteResponseFilter:作用同NettyWriteResponseFilter
  • LoadBalancerClientFilter:网关提供了负载均衡的Filter,具体负载规则可以自己实现
public class LoadBalancerClientFilter implements GlobalFilter, Ordered {
   
	// 可以自己去实现
	protected final LoadBalancerClient loadBalancer;
}

4.2、GatewayFilter

GatewayFilter是面向开发人员的,因需适配,当我们需要给符合predicate的url做一些处理时通过配置就可添加,例如,我们想给path匹配上/test/**的url添加header,通过下列配置就可添加,这类配置是根据业务需求进行的特殊配置。工厂较多,也比较简单,具体配置方法可参考官网 :传送带

 

接口定义

public interface GatewayFilter extends ShortcutConfigurable {
   
     

	/**
	 * Name key.
	 */
	String NAME_KEY = "name";

	/**
	 * Value key.
	 */
	String VALUE_KEY = "value";
	
	Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain);
}

4.3、GatewayFilter对比GlobalFilter

前者接口定义中多了NAME_KEY和VALUE_KEY,原因是GatewayFilter是面向开发人员的,例如我们需要配置给path符合/test/**的请求添加header时,header是key-value形式,这时候就用到了

spring:
  cloud:
    gateway:
      routes:
      - id: add_request_header_route
        uri: http://localhost:8080
        predicates:
        - Path=/test/**
        filters:
        - AddRequestHeader=NAME, test

public class AddRequestHeaderGatewayFilterFactory
		extends AbstractNameValueGatewayFilterFactory {
   
     

	@Override
	public GatewayFilter apply(NameValueConfig config) {
   
     
		return (exchange, chain) -> {
   
     
		// 将要添加的key-value添加到上下文的header中
			ServerHttpRequest request = exchange.getRequest().mutate()
					.header(config.getName(), config.getValue()).build();

			return chain.filter(exchange.mutate().request(request).build());
		};
	}

}

4.4、应用

上述两种Filter作用不同,是在什么时候整合的?优先级如何处理?
每个Filter中都有一个Order属性,在执行时是在FilteringWebHandler#handle方法中对GlobalFilter和GatewayFilter进行的整合和排序,具体执行在FilteringWebHandler#filter方法

/**
* 整合Filter 
*/
public Mono<Void> handle(ServerWebExchange exchange) {
   
     
		// 根据Route信息取出配置的GatewayFilter集合
		Route route = exchange.getRequiredAttribute(GATEWAY_ROUTE_ATTR);
		List<GatewayFilter> gatewayFilters = route.getFilters();
		// 取出globalFilters
		List<GatewayFilter> combined = new ArrayList<>(this.globalFilters);
		// 将GatewayFilter添加到combined
		combined.addAll(gatewayFilters);
		// combined根据Order排优先级
		AnnotationAwareOrderComparator.sort(combined);

		if (logger.isDebugEnabled()) {
   
     
			logger.debug("Sorted gatewayFilterFactories: " + combined);
		}

		return new DefaultGatewayFilterChain(combined).filter(exchange);
}
	
/**
* 执行Filter 
*/
public Mono<Void> filter(ServerWebExchange exchange) {
   
     
			return Mono.defer(() -> {
   
     
				if (this.index < filters.size()) {
   
     
					GatewayFilter filter = filters.get(this.index);
					DefaultGatewayFilterChain chain = new DefaultGatewayFilterChain(this,
							this.index + 1);
					return filter.filter(exchange, chain);
				}
				else {
   
     
					return Mono.empty(); // complete
				}
			});
}

执行过程中,可以看到排完优先级的Filter,其中红框圈出来的是我写的一个Demo,设置的一个GatewayFilter。
 

5、自定义Filter

5.1、自定义GlobalFilter

GlobalFilter具体的实现方式是实现接口,每个filter都实现了GlobalFilter接口

public class GlobalTestFilter implements GlobalFilter, Ordered {
   
     

    @Override
    public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
   
     
        if("符合业务逻辑,处理完业务逻辑,继续执行下一个filter"){
   
     
            return chain.filter(exchange);
        }
        //不符合业务逻辑,直接返回
        return "按照不符合业务逻辑处理";
    }
}

5.2、自定义GatewayFilter

GatewayFilter具体的实现方式是工厂,每个工厂都继承了AbstractGatewayFilterFactory

public class TestGatewayFilterFactory extends AbstractGatewayFilterFactory<TestGatewayFilterFactory.Config> {
   
     
    public TestGatewayFilterFactory() {
   
     
        super(TestGatewayFilterFactory.Config.class);
    }

    @Override
    public GatewayFilter apply(Config config) {
   
     
        return (exchange, chain) -> {
   
     
            if("符合条件,处理业务逻辑,继续执行下一个Filter"){
   
     
                return chain.filter(exchange);
            }
             // 不符合条件,直接返回
            return "false";
        };
    }

    public static class Config {
   
     

        private String businessAttributes;

        public String getBusinessAttributes() {
   
     
            return businessAttributes;
        }

        public void setBusinessAttributes(String businessAttributes) {
   
     
            this.businessAttributes = businessAttributes;
        }
    }
}

6、源码解析

6.1、网关启动阶段

 

①:yaml文件和GatewayProperties文件映射,映射处理源码在JavaBeanBinder.BeanProperty#getValue–>CollectionBinder#merge—>Binder#bindBean
②:加载Locator Bean,为后续读取RouteDefinition做准备【GatewayAutoConfiguration】,下述代码中的Locator就是本文的第二大节介绍的

public class GatewayAutoConfiguration {
   
     
	@Bean
	@ConditionalOnMissingBean
	public PropertiesRouteDefinitionLocator propertiesRouteDefinitionLocator(
			GatewayProperties properties) {
   
     
		return new PropertiesRouteDefinitionLocator(properties);
	}

	@Bean
	@ConditionalOnMissingBean(RouteDefinitionRepository.class)
	public InMemoryRouteDefinitionRepository inMemoryRouteDefinitionRepository() {
   
     
		return new InMemoryRouteDefinitionRepository();
	}

	@Bean
	@Primary
	public RouteDefinitionLocator routeDefinitionLocator(
			List<RouteDefinitionLocator> routeDefinitionLocators) {
   
     
		return new CompositeRouteDefinitionLocator(
				Flux.fromIterable(routeDefinitionLocators));
	}
	
	@Bean
	@Primary
	// TODO: property to disable composite?
	public RouteLocator cachedCompositeRouteLocator(List<RouteLocator> routeLocators) {
   
     
		return new CachingRouteLocator(
				new CompositeRouteLocator(Flux.fromIterable(routeLocators)));
	}
}

③:初始化GlobalFilters【FilteringWebHandler】

public class GatewayAutoConfiguration {
   
     
	@Bean
	public FilteringWebHandler filteringWebHandler(List<GlobalFilter> globalFilters) {
   
     
		return new FilteringWebHandler(globalFilters);
	}
}

public class FilteringWebHandler implements WebHandler {
   
     

	private final List<GatewayFilter> globalFilters;
	//构造函数中设置globalFiltersglobalFilters
	public FilteringWebHandler(List<GlobalFilter> globalFilters) {
   
     
		this.globalFilters = loadFilters(globalFilters);
	}
	//设置globalFilters
	private static List<GatewayFilter> loadFilters(List<GlobalFilter> filters) {
   
     
		return filters.stream().map(filter -> {
   
     
			GatewayFilterAdapter gatewayFilter = new GatewayFilterAdapter(filter);
			if (filter instanceof Ordered) {
   
     
				int order = ((Ordered) filter).getOrder();
				return new OrderedGatewayFilter(gatewayFilter, order);
			}
			return gatewayFilter;
		}).collect(Collectors.toList());
	}
}

④:初始化predicates,gatewayFilters,getRoutes【GatewayAutoConfiguration–>RouteDefinitionRouteLocator】

public class RouteDefinitionRouteLocator
		implements RouteLocator, BeanFactoryAware, ApplicationEventPublisherAware {
   
     
	//构造函数中初始化
	public RouteDefinitionRouteLocator(RouteDefinitionLocator routeDefinitionLocator,
			List<RoutePredicateFactory> predicates,
			List<GatewayFilterFactory> gatewayFilterFactories,
			GatewayProperties gatewayProperties, ConversionService conversionService) {
   
     
		this.routeDefinitionLocator = routeDefinitionLocator;
		this.conversionService = conversionService;
		initFactories(predicates);
		gatewayFilterFactories.forEach(
				factory -> this.gatewayFilterFactories.put(factory.name(), factory));
		this.gatewayProperties = gatewayProperties;
	}
	
	//设置predicate工厂
	private void initFactories(List<RoutePredicateFactory> predicates) {
   
     
		predicates.forEach(factory -> {
   
     
			String key = factory.name();
			if (this.predicates.containsKey(key)) {
   
     
				this.logger.warn("A RoutePredicateFactory named " + key
						+ " already exists, class: " + this.predicates.get(key)
						+ ". It will be overwritten.");
			}
			this.predicates.put(key, factory);
			if (logger.isInfoEnabled()) {
   
     
				logger.info("Loaded RoutePredicateFactory [" + key + "]");
			}
		});
	}

	public Flux<Route> getRoutes() {
   
     
		//从RouteDefinitions转换为Route,转换过程在convertToRoute方法中实现
		return this.routeDefinitionLocator.getRouteDefinitions().map(this::convertToRoute)
				.map(route -> {
   
     
					if (logger.isDebugEnabled()) {
   
     
						logger.debug("RouteDefinition matched: " + route.getId());
					}
					return route;
				});
	}
	
	//RouteDefinition到Route的转换
	private Route convertToRoute(RouteDefinition routeDefinition) {
   
     
		//从routeDefinition获取predicate
		AsyncPredicate<ServerWebExchange> predicate = combinePredicates(routeDefinition);
		//从routeDefinition获取gatewayFilters
		List<GatewayFilter> gatewayFilters = getFilters(routeDefinition);
		//构造Route
		return Route.async(routeDefinition).asyncPredicate(predicate)
				.replaceFilters(gatewayFilters).build();
	}
	
	//获取GatewayFilters
	private List<GatewayFilter> getFilters(RouteDefinition routeDefinition) {
   
     
		List<GatewayFilter> filters = new ArrayList<>();
		//如果默认filter不为空,则去加载
		if (!this.gatewayProperties.getDefaultFilters().isEmpty()) {
   
     
			filters.addAll(loadGatewayFilters(DEFAULT_FILTERS,
					this.gatewayProperties.getDefaultFilters()));
		}
		//如果Filter不为空,则
		if (!routeDefinition.getFilters().isEmpty()) {
   
     
			filters.addAll(loadGatewayFilters(routeDefinition.getId(),
					routeDefinition.getFilters()));
		}

		AnnotationAwareOrderComparator.sort(filters);
		return filters;
	}
	
	@SuppressWarnings("unchecked")
	private List<GatewayFilter> loadGatewayFilters(String id,
			List<FilterDefinition> filterDefinitions) {
   
     
		List<GatewayFilter> filters = filterDefinitions.stream().map(definition -> {
   
     
			//从gatewayFilterFactories中根据key获取factory
			GatewayFilterFactory factory = this.gatewayFilterFactories
					.get(definition.getName());
			if (factory == null) {
   
     
				throw new IllegalArgumentException(
						"Unable to find GatewayFilterFactory with name "
								+ definition.getName());
			}
			//获取definition设置的Filter值
			Map<String, String> args = definition.getArgs();
			if (logger.isDebugEnabled()) {
   
     
				logger.debug("RouteDefinition " + id + " applying filter " + args + " to "
						+ definition.getName());
			}

			Map<String, Object> properties = factory.shortcutType().normalize(args,
					factory, this.parser, this.beanFactory);
			//每一个工厂中都有一个静态内部类Config,目的是存储我们设置的Filter值
			Object configuration = factory.newConfig();
			//将后几个参数的信息绑定到configuration
			ConfigurationUtils.bind(configuration, properties,
					factory.shortcutFieldPrefix(), definition.getName(), validator,
					conversionService);
			//获得GatewayFilter
			GatewayFilter gatewayFilter = factory.apply(configuration);
			if (this.publisher != null) {
   
     
				this.publisher.publishEvent(new FilterArgsEvent(this, id, properties));
			}
			return gatewayFilter;
		}).collect(Collectors.toList());

		ArrayList<GatewayFilter> ordered = new ArrayList<>(filters.size());
		for (int i = 0; i < filters.size(); i++) {
   
     
			GatewayFilter gatewayFilter = filters.get(i);
			if (gatewayFilter instanceof Ordered) {
   
     
				ordered.add(gatewayFilter);
			}
			else {
   
     
				ordered.add(new OrderedGatewayFilter(gatewayFilter, i + 1));
			}
		}

		return ordered;
	}
}

6.2、请求处理阶段

 

①ReactorHttpHandlerAdapter#apply方法是请求到网关执行的入口

public class ReactorHttpHandlerAdapter implements BiFunction<HttpServerRequest, HttpServerResponse, Mono<Void>> {
   
     
	public Mono<Void> apply(HttpServerRequest reactorRequest, HttpServerResponse reactorResponse) {
   
     
		NettyDataBufferFactory bufferFactory = new NettyDataBufferFactory(reactorResponse.alloc());
		try {
   
     
			//获取请求的request和response
			ReactorServerHttpRequest request = new ReactorServerHttpRequest(reactorRequest, bufferFactory);
			ServerHttpResponse response = new ReactorServerHttpResponse(reactorResponse, bufferFactory);

			if (request.getMethod() == HttpMethod.HEAD) {
   
     
				response = new HttpHeadResponseDecorator(response);
			}
			//给到HttpWebHandlerAdapter执行构建
			return this.httpHandler.handle(request, response)
					.doOnError(ex -> logger.trace(request.getLogPrefix() + "Failed to complete: " + ex.getMessage()))
					.doOnSuccess(aVoid -> logger.trace(request.getLogPrefix() + "Handling completed"));
		}
		catch (URISyntaxException ex) {
   
     
			if (logger.isDebugEnabled()) {
   
     
				logger.debug("Failed to get request URI: " + ex.getMessage());
			}
			reactorResponse.status(HttpResponseStatus.BAD_REQUEST);
			return Mono.empty();
		}
	}
}

②HttpWebHandlerAdapter#handle构建网关上下文ServerWebExchange

public class HttpWebHandlerAdapter extends WebHandlerDecorator implements HttpHandler {
   
     
	public Mono<Void> handle(ServerHttpRequest request, ServerHttpResponse response) {
   
     

		if (this.forwardedHeaderTransformer != null) {
   
     
			request = this.forwardedHeaderTransformer.apply(request);
		}
		//根据请求的request、response构建网关上下文
		ServerWebExchange exchange = createExchange(request, response);

		LogFormatUtils.traceDebug(logger, traceOn ->
				exchange.getLogPrefix() + formatRequest(exchange.getRequest()) +
						(traceOn ? ", headers=" + formatHeaders(exchange.getRequest().getHeaders()) : ""));

		return getDelegate().handle(exchange)
				.doOnSuccess(aVoid -> logResponse(exchange))
				.onErrorResume(ex -> handleUnresolvedError(exchange, ex))
				.then(Mono.defer(response::setComplete));
	}
}

③DispatcherHandler用于Http请求处理器/控制器的中央分发处理器,把请求分发给已经注册的处理程序处理,DispatcherHandler遍历Mapping获取对应的handler,网关一共有6个handlerMapping【此处会找到RoutePredicateHandlerMapping,通过RoutePredicateHandlerMapping获取FilteringWebHandler,通过FilteringWebHandler获取】

public class DispatcherHandler implements WebHandler, ApplicationContextAware {
   
     
	public Mono<Void> handle(ServerWebExchange exchange) {
   
     
		if (this.handlerMappings == null) {
   
     
			return createNotFoundError();
		}
		//遍历mapping获取handler
		return Flux.fromIterable(this.handlerMappings)
				.concatMap(mapping -> mapping.getHandler(exchange))
				.next()
				.switchIfEmpty(createNotFoundError())
				.flatMap(handler -> invokeHandler(exchange, handler))
				.flatMap(result -> handleResult(exchange, result));
	}
}

public class RoutePredicateHandlerMapping extends AbstractHandlerMapping {
   
     

	private final FilteringWebHandler webHandler;

	private final RouteLocator routeLocator;

	private final Integer managementPort;

	private final ManagementPortType managementPortType;
	//网关启动时进行了初始化
	public RoutePredicateHandlerMapping(FilteringWebHandler webHandler,
			RouteLocator routeLocator, GlobalCorsProperties globalCorsProperties,
			Environment environment) {
   
     
		this.webHandler = webHandler;
		this.routeLocator = routeLocator;

		this.managementPort = getPortProperty(environment, "management.server.");
		this.managementPortType = getManagementPortType(environment);
		setOrder(1);
		setCorsConfigurations(globalCorsProperties.getCorsConfigurations());
	}

	protected Mono<?> getHandlerInternal(ServerWebExchange exchange) {
   
     
		// don't handle requests on management port if set and different than server port
		if (this.managementPortType == DIFFERENT && this.managementPort != null
				&& exchange.getRequest().getURI().getPort() == this.managementPort) {
   
     
			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);
					//返回FilteringWebHandler
					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) + "]");
					}
				})));
	}
}

④ RoutePredicateHandlerMapping#lookupRoute匹配路由,根据routeLocator获取我们在配置我文件中配置的Route,和当前请求的路由做匹配

public class RoutePredicateHandlerMapping extends AbstractHandlerMapping {
   
     
	protected Mono<Route> lookupRoute(ServerWebExchange exchange) {
   
     
		//routeLocator获取我们在配置我文件中配置的Route
		return this.routeLocator.getRoutes()
				.concatMap(route -> Mono.just(route).filterWhen(r -> {
   
     
					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 -> {
   
     
					if (logger.isDebugEnabled()) {
   
     
						logger.debug("Route matched: " + route.getId());
					}
					validateRoute(route, exchange);
					return route;
				});
	}
}

⑤FilteringWebHandler创建过滤器链,执行过滤器

public class FilteringWebHandler implements WebHandler {
   
     
	//创建过滤器链
	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);
	}

	private static class DefaultGatewayFilterChain implements GatewayFilterChain {
   
     
		//调用过滤器
		public Mono<Void> filter(ServerWebExchange exchange) {
   
     
			return Mono.defer(() -> {
   
     
				if (this.index < filters.size()) {
   
     
					GatewayFilter filter = filters.get(this.index);
					DefaultGatewayFilterChain chain = new DefaultGatewayFilterChain(this,
							this.index + 1);
					//执行调用
					return filter.filter(exchange, chain);
				}
				else {
   
     
					return Mono.empty(); // complete
				}
			});
		}
	}
}

版权声明:「DDKK.COM 弟弟快看,程序员编程资料站」本站文章,版权归原作者所有