環境：SpringBoot2.7.10 + Spring Cloud gateway3.1.6
1 RouteToRequestUrlFilter
根據路由配置的url信息，構建成為要訪問的目標地址，如下路由配置：?
spring: cloud: gateway: enabled: true # 全局超時配置 httpclient: connect-timeout: 10000 response-timeout: 5000 discovery: locator: enabled: true lowerCaseServiceId: true # 這里是全局過濾器，也就是下面在介紹過濾器執行的時候一定會執行StripPrefixGatewayFilterFactory#apply # 返回的過濾器，如下路由配置：該過濾器會將你的請求轉換為：http://localhost:8088/demos，保存到上下文中 # ServerWebExchange#getAttributes().put(GATEWAY_REQUEST_URL_ATTR, newRequest.getURI()) default-filters: - StripPrefix=1 routes: - id: R001 uri: http://localhost:8787 predicates: - Path=/api-1/**,/api-2/** metadata: akf: "dbc" #局部超時設置 connect-timeout: 10000 response-timeout: 5000 - id: st001 uri: lb://storage-service predicates: - Path=/api-x/** - id: o001 uri: lb://order-service predicates: - Path=/api-a/**, /api-b/** metadata: akf: "dbc" #局部超時設置 connect-timeout: 10000 response-timeout: 5000
訪問：??http://localhost:8088/api-1/demos??
(資料圖片僅供參考)
轉換后：??http://localhost:8787/demos??
該過濾器最后會將轉換后的url保存到上下文中
ServerWebExchange#getAttributes().put(GATEWAY_REQUEST_URL_ATTR, mergedUrl);
注意：上面的StripPrefixGatewayFilterFactory#apply過濾器執行完后，才會執行該過濾器。
總結：
訪問：http://localhost:9090/api-x/orders ，路由地址：lb://order-service
轉換地址轉換后：http://localhost:9090/orders合并地址將上一步的地址進一步合并為：lb://order-service/orders將地址存儲到上下文中：exchange.getAttributes().put(GATEWAY_REQUEST_URL_ATTR, mergedUrl);2 ReactiveLoadBalancerClientFilter
如果URL有一個lb(例如lb://order-service)，它使用Spring Cloud ReactorLoadBalancer將名稱(在本例中為order-service)解析為一個實際的主機和端口，并替換相同屬性中的URI。?
public class ReactiveLoadBalancerClientFilter implements GlobalFilter, Ordered { private final LoadBalancerClientFactory clientFactory; private final GatewayLoadBalancerProperties properties; private final LoadBalancerProperties loadBalancerProperties; public Mono filter(ServerWebExchange exchange, GatewayFilterChain chain) { // 從上下文中獲取，如：lb://order-service/orders URI url = exchange.getAttribute(GATEWAY_REQUEST_URL_ATTR); String schemePrefix = exchange.getAttribute(GATEWAY_SCHEME_PREFIX_ATTR); if (url == null || (!"lb".equals(url.getScheme()) && !"lb".equals(schemePrefix))) { return chain.filter(exchange); } // preserve the original url addOriginalRequestUrl(exchange, url); // 再次獲取 URI requestUri = exchange.getAttribute(GATEWAY_REQUEST_URL_ATTR); // 獲取服務名;order-service String serviceId = requestUri.getHost(); // clientFactory.getInstances方法會從NamedContextFactory.contexts集合中查找以order-service為key對應的 // AnnotationConfigApplicationContext，然后從這個容器中查找LoadBalancerLifecycle，默認返回{} // ------------------------------------------------------------ /** * 每個服務對應的ApplicationContext包含如下13個Bean * org.springframework.context.annotation.internalConfigurationAnnotationProcessor * org.springframework.context.annotation.internalAutowiredAnnotationProcessor * org.springframework.context.annotation.internalCommonAnnotationProcessor * org.springframework.context.event.internalEventListenerProcessor * org.springframework.context.event.internalEventListenerFactory * propertyPlaceholderAutoConfiguration loadBalancerClientConfiguration * propertySourcesPlaceholderConfigurer * LoadBalancerClientConfiguration$ReactiveSupportConfiguration * discoveryClientServiceInstanceListSupplier * LoadBalancerClientConfiguration$BlockingSupportConfiguration, * reactorServiceInstanceLoadBalancer */ // 這里集合返回{} Set supportedLifecycleProcessors = LoadBalancerLifecycleValidator .getSupportedLifecycleProcessors(clientFactory.getInstances(serviceId, LoadBalancerLifecycle.class), RequestDataContext.class, ResponseData.class, ServiceInstance.class); DefaultRequest lbRequest = new DefaultRequest<>(new RequestDataContext( new RequestData(exchange.getRequest()), getHint(serviceId, loadBalancerProperties.getHint()))); // choose負載查找指定服務(order-server) return choose(lbRequest, serviceId, supportedLifecycleProcessors).doOnNext(response -> { if (!response.hasServer()) { supportedLifecycleProcessors.forEach(lifecycle -> lifecycle .onComplete(new CompletionContext<>(CompletionContext.Status.DISCARD, lbRequest, response))); throw NotFoundException.create(properties.isUse404(), "Unable to find instance for " + url.getHost()); } ServiceInstance retrievedInstance = response.getServer(); URI uri = exchange.getRequest().getURI(); // if the `lb:` mechanism was used, use `` as the default, // if the loadbalancer doesn"t provide one. String overrideScheme = retrievedInstance.isSecure() ? "https" : "http"; if (schemePrefix != null) { overrideScheme = url.getScheme(); } DelegatingServiceInstance serviceInstance = new DelegatingServiceInstance(retrievedInstance, overrideScheme); URI requestUrl = reconstructURI(serviceInstance, uri); exchange.getAttributes().put(GATEWAY_REQUEST_URL_ATTR, requestUrl); exchange.getAttributes().put(GATEWAY_LOADBALANCER_RESPONSE_ATTR, response); supportedLifecycleProcessors.forEach(lifecycle -> lifecycle.onStartRequest(lbRequest, response)); }).then(chain.filter(exchange)) .doOnError(throwable -> supportedLifecycleProcessors.forEach(lifecycle -> lifecycle.onComplete( new CompletionContext(CompletionContext.Status.FAILED, throwable, lbRequest, exchange.getAttribute(GATEWAY_LOADBALANCER_RESPONSE_ATTR))))) .doOnSuccess(aVoid -> supportedLifecycleProcessors.forEach( lifecycle -> lifecycle.onComplete(new CompletionContext( CompletionContext.Status.SUCCESS, lbRequest, exchange.getAttribute(GATEWAY_LOADBALANCER_RESPONSE_ATTR), new ResponseData(exchange.getResponse(), new RequestData(exchange.getRequest())))))); } protected URI reconstructURI(ServiceInstance serviceInstance, URI original) { return LoadBalancerUriTools.reconstructURI(serviceInstance, original); } private Mono> choose(Request lbRequest, String serviceId, Set supportedLifecycleProcessors) { // 從order-service對應的ApplicationContext中查找ReactorServiceInstanceLoadBalancer ReactorLoadBalancer loadBalancer = this.clientFactory.getInstance(serviceId, ReactorServiceInstanceLoadBalancer.class); if (loadBalancer == null) { throw new NotFoundException("No loadbalancer available for " + serviceId); } supportedLifecycleProcessors.forEach(lifecycle -> lifecycle.onStart(lbRequest)); // 查找服務實例 return loadBalancer.choose(lbRequest); } private String getHint(String serviceId, Map hints) { String defaultHint = hints.getOrDefault("default", "default"); String hintPropertyValue = hints.get(serviceId); return hintPropertyValue != null ? hintPropertyValue : defaultHint; }}// 輪詢算分public class RoundRobinLoadBalancer implements ReactorServiceInstanceLoadBalancer { final AtomicInteger position; ObjectProvider serviceInstanceListSupplierProvider; public Mono> choose(Request request) { // 接下面ClientFactoryObjectProvider中獲取ServiceInstanceListSupplier ServiceInstanceListSupplier supplier = serviceInstanceListSupplierProvider .getIfAvailable(NoopServiceInstanceListSupplier::new); return supplier.get(request).next().map(serviceInstances -> processInstanceResponse(supplier, serviceInstances)); } private Response processInstanceResponse(ServiceInstanceListSupplier supplier, List serviceInstances) { Response serviceInstanceResponse = getInstanceResponse(serviceInstances); if (supplier instanceof SelectedInstanceCallback && serviceInstanceResponse.hasServer()) { ((SelectedInstanceCallback) supplier).selectedServiceInstance(serviceInstanceResponse.getServer()); } return serviceInstanceResponse; } private Response getInstanceResponse(List instances) { if (instances.isEmpty()) { return new EmptyResponse(); } // TODO: enforce order? int pos = Math.abs(this.position.incrementAndGet()); ServiceInstance instance = instances.get(pos % instances.size()); return new DefaultResponse(instance); }}class ClientFactoryObjectProvider implements ObjectProvider { private final NamedContextFactory clientFactory; // type = ServiceInstanceListSupplier private final Class type; // name = order-service private final String name; private ObjectProvider delegate() { if (this.provider == null) { // 從order-service對應ApplicationContext中獲取ServiceInstanceListSupplier // 這里最終返回的是：DiscoveryClientServiceInstanceListSupplier this.provider = this.clientFactory.getProvider(this.name, this.type); } return this.provider; }}public class LoadBalancerClientConfiguration { @Configuration(proxyBeanMethods = false) @ConditionalOnReactiveDiscoveryEnabled @Order(REACTIVE_SERVICE_INSTANCE_SUPPLIER_ORDER) public static class ReactiveSupportConfiguration { @Bean @ConditionalOnBean(ReactiveDiscoveryClient.class) @ConditionalOnMissingBean @ConditionalOnProperty(value = "spring.cloud.loadbalancer.configurations", havingValue = "default", matchIfMissing = true) public ServiceInstanceListSupplier discoveryClientServiceInstanceListSupplier( ConfigurableApplicationContext context) { // 這里最終構建的是：DiscoveryClientServiceInstanceListSupplier return ServiceInstanceListSupplier.builder().withDiscoveryClient().withCaching().build(context); } }}public final class ServiceInstanceListSupplierBuilder { public ServiceInstanceListSupplierBuilder withDiscoveryClient() { this.baseCreator = context -> { // 先從order-service對應的ApplicationContext中查找ReactiveDiscoveryClient，如果你沒有自定義，那么就會從 // 父容器中查找，如果你使用的nacos，那么會返回NacosReactiveDiscoveryClient ReactiveDiscoveryClient discoveryClient = context.getBean(ReactiveDiscoveryClient.class); return new DiscoveryClientServiceInstanceListSupplier(discoveryClient, context.getEnvironment()); }; return this; }}總結：
獲取地址獲取上一步中保存在上下文的地址URI url = exchange.getAttribute(GATEWAY_REQUEST_URL_ATTR);獲取LoadBalancerLifecycle取得當前服務(order-service)，對應的AnnotationConfigApplicationContext中配置的LoadBalancerLifecycle，該負載均衡生命周期能夠監控負載均衡的執行過程。該類是泛型類，3個泛型參數，類型依次為：RequestDataContext.class, ResponseData.class, ServiceInstance.class。獲取ReactorServiceInstanceLoadBalancer獲取當前服務(order-server)，對應的AnnotationConfigApplicationContext中配置的ReactorServiceInstanceLoadBalancer。每一個服務都有一個對應的默認配置類LoadBalancerClientConfiguration，該配置類中有默認的RoundRobinLoadBalancer。我們可以為具體的服務提供LoadBalancerClientSpecification類型的Bean，該類我們可以指定你要配置的serviceId及配置類，在配置類中我們可以自定義ReactorServiceInstanceLoadBalancer的實現類Bean。選擇服務在上一步中獲得ReactorServiceInstanceLoadBalancer后，接下來就是選取一個服務實例了。重構URI上一步中獲取了ServiceInstance后就能夠重構URL了，當前的URL為: http://localhost:9090/orders 構建后：http://localhost:9093/storages ，將該URL保存到上下文中 exchange.getAttributes().put(GATEWAY_REQUEST_URL_ATTR, requestUrl);3 NettyRoutingFilterpublic class NettyRoutingFilter implements GlobalFilter { private final HttpClient httpClient; public Mono filter(ServerWebExchange exchange, GatewayFilterChain chain) { // 從上下文中獲取解析后的目標地址 URI requestUrl = exchange.getRequiredAttribute(GATEWAY_REQUEST_URL_ATTR); // ... // 獲取上下文中的路由信息 Route route = exchange.getAttribute(GATEWAY_ROUTE_ATTR); // getHttpClient獲取客戶端信息 Flux responseFlux = getHttpClient(route, exchange).headers(headers -> { // ... }).request(method).uri(url).send((req, nettyOutbound) -> { // 發送網絡請求 return nettyOutbound.send(request.getBody().map(this::getByteBuf)); }).responseConnection((res, connection) -> { exchange.getAttributes().put(CLIENT_RESPONSE_ATTR, res); // 建立的Connection對象保存到上下文中，在后續的NettyWriteResponseFilter中會獲取該對象獲取響應數據 exchange.getAttributes().put(CLIENT_RESPONSE_CONN_ATTR, connection); ServerHttpResponse response = exchange.getResponse(); HttpHeaders headers = new HttpHeaders(); res.responseHeaders().forEach(entry -> headers.add(entry.getKey(), entry.getValue())); String contentTypeValue = headers.getFirst(HttpHeaders.CONTENT_TYPE); if (StringUtils.hasLength(contentTypeValue)) { exchange.getAttributes().put(ORIGINAL_RESPONSE_CONTENT_TYPE_ATTR, contentTypeValue); } setResponseStatus(res, response); HttpHeaders filteredResponseHeaders = HttpHeadersFilter.filter(getHeadersFilters(), headers, exchange, Type.RESPONSE); if (!filteredResponseHeaders.containsKey(HttpHeaders.TRANSFER_ENCODING) && filteredResponseHeaders.containsKey(HttpHeaders.CONTENT_LENGTH)) { response.getHeaders().remove(HttpHeaders.TRANSFER_ENCODING); } exchange.getAttributes().put(CLIENT_RESPONSE_HEADER_NAMES, filteredResponseHeaders.keySet()); response.getHeaders().putAll(filteredResponseHeaders); return Mono.just(res); }); // 從路由中的元數據中獲取response-timeout響應超時時間 Duration responseTimeout = getResponseTimeout(route); if (responseTimeout != null) { responseFlux = responseFlux // 設置超時時間 .timeout(responseTimeout, Mono.error(new TimeoutException("Response took longer than timeout: " + responseTimeout))) .onErrorMap(TimeoutException.class, th -> new ResponseStatusException(HttpStatus.GATEWAY_TIMEOUT, th.getMessage(), th)); } return responseFlux.then(chain.filter(exchange)); } protected HttpClient getHttpClient(Route route, ServerWebExchange exchange) { // 從路由的元數據中獲取配置的連接超時時間：connect-timeout Object connectTimeoutAttr = route.getMetadata().get(CONNECT_TIMEOUT_ATTR); if (connectTimeoutAttr != null) { Integer connectTimeout = getInteger(connectTimeoutAttr); // 設置Netty的連接超時時間 // io.netty.channel.ChannelOption return this.httpClient.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, connectTimeout); } return httpClient; }}總結：
獲取URL獲取上一步保存在上下文中的URLURI requestUrl = exchange.getRequiredAttribute(GATEWAY_REQUEST_URL_ATTR);設置當前路由狀態設置當前路由已經路由狀態setAlreadyRouted(exchange);exchange.getAttributes().put(GATEWAY_ALREADY_ROUTED_ATTR, true);獲取路由Route route = exchange.getAttribute(GATEWAY_ROUTE_ATTR);獲取當前的Route信息。主要就用來獲取配置路由時提供的配置信息，比如：超時時間設置，如上配置。RoutePredicateHandlerMapping#getHandlerInternal方法中保存路由到上下文中構建HttpClient通過上一步取得的Route對象，配置HttpClient相關屬性，比如：超時時間。配置基本的http相關信息，建立連接后將Connection對象保存到上下文中，供下一個過濾器獲取響應數據4 NettyWriteResponseFilter該過濾器的作用是處理由NettyRoutingFilter中建立的HTTP請求（包括：請求參數，請求頭，建立連接）；在NettyRoutingFilter中會將建立連接后的Connection保存到ServerWebExchange上下文中。?
public class NettyWriteResponseFilter implements GlobalFilter, Ordered { public Mono filter(ServerWebExchange exchange, GatewayFilterChain chain) { // NOTICE: nothing in "pre" filter stage as CLIENT_RESPONSE_CONN_ATTR is not added // until the NettyRoutingFilter is run // @formatter:off return chain.filter(exchange) .doOnError(throwable -> cleanup(exchange)) .then(Mono.defer(() -> { Connection connection = exchange.getAttribute(CLIENT_RESPONSE_CONN_ATTR); if (connection == null) { return Mono.empty(); } ServerHttpResponse response = exchange.getResponse(); // TODO: needed? final Flux body = connection .inbound() .receive() .retain() .map(byteBuf -> wrap(byteBuf, response)); MediaType contentType = null; try { contentType = response.getHeaders().getContentType(); } // 根據不同的ContentType做不同的響應 return (isStreamingMediaType(contentType) ? response.writeAndFlushWith(body.map(Flux::just)) : response.writeWith(body)); })).doOnCancel(() -> cleanup(exchange)); // @formatter:on } protected DataBuffer wrap(ByteBuf byteBuf, ServerHttpResponse response) { DataBufferFactory bufferFactory = response.bufferFactory(); if (bufferFactory instanceof NettyDataBufferFactory) { NettyDataBufferFactory factory = (NettyDataBufferFactory) bufferFactory; return factory.wrap(byteBuf); } // MockServerHttpResponse creates these else if (bufferFactory instanceof DefaultDataBufferFactory) { DataBuffer buffer = ((DefaultDataBufferFactory) bufferFactory).allocateBuffer(byteBuf.readableBytes()); buffer.write(byteBuf.nioBuffer()); byteBuf.release(); return buffer; } throw new IllegalArgumentException("Unkown DataBufferFactory type " + bufferFactory.getClass()); } private void cleanup(ServerWebExchange exchange) { Connection connection = exchange.getAttribute(CLIENT_RESPONSE_CONN_ATTR); if (connection != null && connection.channel().isActive() && !connection.isPersistent()) { connection.dispose(); } } private boolean isStreamingMediaType(@Nullable MediaType contentType) { return (contentType != null && this.streamingMediaTypes.stream().anyMatch(contentType::isCompatibleWith)); }}總結：
取得Connection取得上一步中保存的ConnectionConnection connection = exchange.getAttribute(CLIENT_RESPONSE_CONN_ATTR);響應內容輸出微服務端響應的數據?final Flux body = connection .inbound() .receive() .retain() .map(byteBuf -> wrap(byteBuf, response));以上就是Gateway在處理一個路由請求的執行流程
完畢?。?！
關鍵詞：
