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

 ServerWebExchange#getAttributes().put(GATEWAY_REQUEST_URL_ATTR, mergedUrl);

public class ReactiveLoadBalancerClientFilter implements GlobalFilter, Ordered {  private final LoadBalancerClientFactory clientFactory;  private final GatewayLoadBalancerProperties properties;  private final LoadBalancerProperties loadBalancerProperties;
  public Mono<Void> 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<LoadBalancerLifecycle> supportedLifecycleProcessors = LoadBalancerLifecycleValidator        .getSupportedLifecycleProcessors(clientFactory.getInstances(serviceId, LoadBalancerLifecycle.class),            RequestDataContext.class, ResponseData.class, ServiceInstance.class);    DefaultRequest<RequestDataContext> 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:<scheme>` mechanism was used, use `<scheme>` 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<ResponseData, ServiceInstance, RequestDataContext>(CompletionContext.Status.FAILED,                throwable, lbRequest, exchange.getAttribute(GATEWAY_LOADBALANCER_RESPONSE_ATTR)))))        .doOnSuccess(aVoid -> supportedLifecycleProcessors.forEach(            lifecycle -> lifecycle.onComplete(new CompletionContext<ResponseData, ServiceInstance, RequestDataContext>(                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<Response<ServiceInstance>> choose(Request<RequestDataContext> lbRequest, String serviceId,      Set<LoadBalancerLifecycle> supportedLifecycleProcessors) {    // 从order-service对应的ApplicationContext中查找ReactorServiceInstanceLoadBalancer    ReactorLoadBalancer<ServiceInstance> 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<String, String> 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<ServiceInstanceListSupplier> serviceInstanceListSupplierProvider;
  public Mono<Response<ServiceInstance>> choose(Request request) {    // 接下面ClientFactoryObjectProvider中获取ServiceInstanceListSupplier    ServiceInstanceListSupplier supplier = serviceInstanceListSupplierProvider        .getIfAvailable(NoopServiceInstanceListSupplier::new);    return supplier.get(request).next().map(serviceInstances -> processInstanceResponse(supplier, serviceInstances));  }
  private Response<ServiceInstance> processInstanceResponse(ServiceInstanceListSupplier supplier,      List<ServiceInstance> serviceInstances) {    Response<ServiceInstance> serviceInstanceResponse = getInstanceResponse(serviceInstances);    if (supplier instanceof SelectedInstanceCallback && serviceInstanceResponse.hasServer()) {      ((SelectedInstanceCallback) supplier).selectedServiceInstance(serviceInstanceResponse.getServer());    }    return serviceInstanceResponse;  }
  private Response<ServiceInstance> getInstanceResponse(List<ServiceInstance> 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<T> implements ObjectProvider<T> {  private final NamedContextFactory<?> clientFactory;  // type = ServiceInstanceListSupplier  private final Class<T> type;  // name = order-service  private final String name;
  private ObjectProvider<T> 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;  }}

public class NettyRoutingFilter implements GlobalFilter {  private final HttpClient httpClient;
  public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {    // 从上下文中获取解析后的目标地址    URI requestUrl = exchange.getRequiredAttribute(GATEWAY_REQUEST_URL_ATTR);    // ...    // 获取上下文中的路由信息    Route route = exchange.getAttribute(GATEWAY_ROUTE_ATTR);    // getHttpClient获取客户端信息    Flux<HttpClientResponse> 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;  }}

public class NettyWriteResponseFilter implements GlobalFilter, Ordered {  public Mono<Void> 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<DataBuffer> 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));  }}

final Flux<DataBuffer> body = connection  .inbound()  .receive()  .retain()  .map(byteBuf -> wrap(byteBuf, response));