SpringCloud网关组件Gateway原理深度解析

spring:
  cloud:
    gateway:
      routes:
      - id: test_route
        uri: lb://service-A
        predicates:
         - Path=/hello
        filters:
        - SetRequestHeader=X-Request-Red, Blue

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

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

spring:
  cloud:
    gateway:
      routes:
      - id: test_route
        uri: lb://service-A
        predicates:
         - Path=/hello
        filters:
        - SetRequestHeader=X-Request-Red, Blue

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

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

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

public interface GatewayFilter extends ShortcutConfigurable {
	/**
	 * Name key.
	 */
	String NAME_KEY = "name";
	/**
	 * Value key.
	 */
	String VALUE_KEY = "value";
	Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain);
}

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

    /**
    * 整合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
                    }
            });
    }

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

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

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
	public RouteLocator cachedCompositeRouteLocator(List<RouteLocator> routeLocators) {
		return new CachingRouteLocator(
				new CompositeRouteLocator(Flux.fromIterable(routeLocators)));
	}
}

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

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

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

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

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) + "]");
					}
				})));
	}
}

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

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