提供了编程的基础技术教程

zl程序教程

您现在的位置是：首页 >  Java

当前栏目

关于Spring Cloud Gateway与下游服务器的连接分析

2023-02-18 16:32:34 时间

背景

最近面试了不少同学，有很大一部分简历上会提到网关，我一般都会顺着往下问他们的网关是怎么做的。

基本上都是说直接使用的Spring Cloud Gateway或者基于Spring Cloud Gateway二次开发。

这种时候我会继续问一个比较基础的问题：Spring Cloud Gateway作为网关，会把接收到的请求转发给下游服务，那么Spring Cloud Gateway跟下游的服务之间保持的是长连还是短连？还是说每次转发的时候都会新建立一个连接吗?

很遗憾的是，这么基础的问题，很少有面试者完全搞清楚。

所以才有了这篇文章：通过研究Spring Cloud Gateway的源码，来看看Spring Cloud Gateway跟下游服务之间是怎么通信的。

Spring Cloud Gateway

在源码分析之前，需要先了解一下Spring Cloud Gateway

SpringCloud Gateway 是 Spring Cloud 的一个全新项目，该项目是基于 Spring 5.0，Spring Boot 2.0 和 Project Reactor 等技术开发的网关，它旨在为微服务架构提供一种简单有效的统一的 API 路由管理方式。

Spring Cloud Gateway是基于Spring WebFlux框架实现的，而WebFlux框架底层则使用了高性能的Reactor模式通信框架Netty。

Spring Cloud Gateway架构图如下：

源码分析

对于基于webflux的应用，入口点都在DispatchHandler.handle()方法:

最终执行到 SimpleHandlerAdapter.handle() 方法

handler()方法中执行的是 FilteringWebHandler.handle()方法

FilteringWebHandler.handler()方法的主要逻辑就是依次执行已经形成的全局过滤器globalFilter的filter()方法。

从截图中可以看到，默认会生成9个全局过滤器GatewayFilter对象。

单步调试下去，发现涉及到网络这一块的操作都在倒数第二个过滤器NettyRoutingFilter类中。

现在着重来看一下NettyRoutingFilter.filter()方法:

public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
    URI requestUrl = exchange.getRequiredAttribute(GATEWAY_REQUEST_URL_ATTR);

    // ... 一些省略代码
    // 获取httpclient
    Flux<HttpClientResponse> responseFlux = getHttpClient(route, exchange)
            .headers(headers -> {
                headers.add(httpHeaders);
                // Will either be set below, or later by Netty
                headers.remove(HttpHeaders.HOST);
                if (preserveHost) {
                    String host = request.getHeaders().getFirst(HttpHeaders.HOST);
                    headers.add(HttpHeaders.HOST, host);
                }
            }).request(method).uri(url).send((req, nettyOutbound) -> {
                if (log.isTraceEnabled()) {
                    nettyOutbound
                            .withConnection(connection -> log.trace("outbound route: "
                                    + connection.channel().id().asShortText()
                                    + ", inbound: " + exchange.getLogPrefix()));
                }
                // 发送请求
                return nettyOutbound.send(request.getBody().map(this::getByteBuf));
            }).responseConnection((res, connection) -> {

                // 省略代码，下游请求返回之后做的一些处理
                return Mono.just(res);
            });

    Duration responseTimeout = getResponseTimeout(route);

    // 一些省略代码
    return responseFlux.then(chain.filter(exchange));
}

上面代码的逻辑主要就是

  1. 获取通信用的httpclient
  2. 设置headers，method和url
  3. 执行responseConnection()方法发起连接
  4. 连接成功之后执行send()方法传入的lambda方法。
  5. 执行responseConnection()传入的lambda方法。

首先来看一下getHttpClient()方法

protected HttpClient getHttpClient(Route route, ServerWebExchange exchange) {
    // 省略代码，timeout设置
    return httpClient;
}

实际上就是直接返回httpClient对象，那么httpClient是在哪里设置的呢？

public NettyRoutingFilter(HttpClient httpClient,
        ObjectProvider<List<HttpHeadersFilter>> headersFiltersProvider,
        HttpClientProperties properties) {
    this.httpClient = httpClient;
    this.headersFiltersProvider = headersFiltersProvider;
    this.properties = properties;
}

可以看到是在生成NettyRoutingFilter对象的时候传入的，那么NettyRoutingFilter对象在哪里生成的呢？

答：在GatewayAutoConfiguration类中生成的，这个类是在引入网关的依赖之后自动引入的。

同样的，HttpClient对象也是在这个类里面生成的。

@Bean
@ConditionalOnMissingBean
public HttpClient gatewayHttpClient(HttpClientProperties properties,
        List<HttpClientCustomizer> customizers) {

    // 配置连接池
    HttpClientProperties.Pool pool = properties.getPool();

    ConnectionProvider connectionProvider;
    if (pool.getType() == DISABLED) {
        connectionProvider = ConnectionProvider.newConnection();
    }
    else if (pool.getType() == FIXED) {
        connectionProvider = ConnectionProvider.fixed(pool.getName(),
                pool.getMaxConnections(), pool.getAcquireTimeout(),
                pool.getMaxIdleTime(), pool.getMaxLifeTime());
    }
    else {
        connectionProvider = ConnectionProvider.elastic(pool.getName(),
                pool.getMaxIdleTime(), pool.getMaxLifeTime());
    }

    HttpClient httpClient = HttpClient.create(connectionProvider)
            // TODO: move customizations to HttpClientCustomizers
            .httpResponseDecoder(spec -> {
                // 省略代码
                return spec;
            }).tcpConfiguration(tcpClient -> {

                // 省略代码
                return tcpClient;
            });

    // 省略代码  ssl设置


    return httpClient;
}

从上面代码可以看出，HttpClient对象自带一个连接池，生成Httpclient的时候首先会配置这个连接池。

可以看到HttpClient提供的连接池的类型：

public enum PoolType {

    /**
     * 弹性的连接池
     */
    ELASTIC,

    /**
     * 固定长度的连接池
     */
    FIXED,

    /**
     * 不使用连接池
     */
    DISABLED

}

默认使用的是第一种 弹性的连接池

private PoolType type = PoolType.ELASTIC;
connectionProvider = ConnectionProvider.elastic(pool.getName(),
						pool.getMaxIdleTime(), pool.getMaxLifeTime());
static ConnectionProvider elastic(String name, @Nullable Duration maxIdleTime, @Nullable Duration maxLifeTime) {
    return builder(name).maxConnections(Integer.MAX_VALUE) //设置最大连接数无限制
                        .pendingAcquireTimeout(Duration.ofMillis(0))
                        .pendingAcquireMaxCount(-1)
                        .maxIdleTime(maxIdleTime)
                        .maxLifeTime(maxLifeTime)
                        .build();
}
static Builder builder(String name) {
    return new Builder(name);
}

在Builder()构造函数中会调用ConnectionPoolSpec()方法:

private ConnectionPoolSpec() {
    if (DEFAULT_POOL_MAX_IDLE_TIME > -1) {
        maxIdleTime(Duration.ofMillis(DEFAULT_POOL_MAX_IDLE_TIME));
    }
    // 支持不同类型的链接保存方式
    // lifo和fifo
    if(LEASING_STRATEGY_LIFO.equals(DEFAULT_POOL_LEASING_STRATEGY)) {
        lifo();
    }
    else {
        fifo();
    }
}

从代码里面可以看到，httpclient自带的连接池还支持两种连接获取方式： lifo(后进先出)和fifo(先进先出) 默认使用的是fifo。

先总结一下，在引入网关的依赖之后，会自动创建一个HttpClient对象，而这个HttpClient对象自带一个连接池，且默认是Elastic连接池，即连接池内的数量会弹性发生变化。 连接池内部默认采用fifo的方式来保存以及使用连接

现在重新回到NettyRoutingFilter.filter()方法中来看下:

public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
    URI requestUrl = exchange.getRequiredAttribute(GATEWAY_REQUEST_URL_ATTR);

    // ... 一些省略代码
    // 获取httpclient
    Flux<HttpClientResponse> responseFlux = getHttpClient(route, exchange)
            .headers(headers -> {
                headers.add(httpHeaders);
                // Will either be set below, or later by Netty
                headers.remove(HttpHeaders.HOST);
                if (preserveHost) {
                    String host = request.getHeaders().getFirst(HttpHeaders.HOST);
                    headers.add(HttpHeaders.HOST, host);
                }
            }).request(method).uri(url).send((req, nettyOutbound) -> {
                if (log.isTraceEnabled()) {
                    nettyOutbound
                            .withConnection(connection -> log.trace("outbound route: "
                                    + connection.channel().id().asShortText()
                                    + ", inbound: " + exchange.getLogPrefix()));
                }
                // 发送请求
                return nettyOutbound.send(request.getBody().map(this::getByteBuf));
            }).responseConnection((res, connection) -> {

                // 省略代码，下游请求返回之后做的一些处理
                return Mono.just(res);
            });

    Duration responseTimeout = getResponseTimeout(route);

    // 一些省略代码
    return responseFlux.then(chain.filter(exchange));
}

responseConnection()方法中会发起连接操作:

final TcpClient cachedConfiguration;

@SuppressWarnings("unchecked")
Mono<HttpClientOperations> connect() {
    return (Mono<HttpClientOperations>)cachedConfiguration.connect();
}

@Override
public <V> Flux<V> responseConnection(BiFunction<? super HttpClientResponse, ? super Connection, ? extends Publisher<V>> receiver) {
    return connect().flatMapMany(resp -> Flux.from(receiver.apply(resp, resp)));
}

调用的是TcpClient对象的connect()方法，一步步断点下去发现最终调用的是TcpClientConnect.connect()方法.

final ConnectionProvider provider;

@Override
public Mono<? extends Connection> connect(Bootstrap b) {

    if (b.config()
         .group() == null) {

        TcpClientRunOn.configure(b,
                LoopResources.DEFAULT_NATIVE,
                TcpResources.get());
    }

    // 这里的provider实际上就是前面分析的创建HttpClient的时候生成的ConnectProvider对象
    return provider.acquire(b);

}

从代码实现中可以看到，实际上TcpClienConnect是直接从ConnectionProvider获取连接。

看到这里，本文一开始的问题其实已经有解答了：

默认情况下(除非显示设置不使用连接池)，网关在把请求转发给下游服务器的时候，是会使用连接池的，而不是每次都重新发起连接。

继续往下分析。

对于Elastic类型的连接池来说，其默认实现为PooledConnectionProvider

// key为远程地址(一般指代一个远程服务),value则对应的ConnectioAllocator
final ConcurrentMap<PoolKey, InstrumentedPool<PooledConnection>> channelPools =
			PlatformDependent.newConcurrentHashMap();

@Override
public Mono<Connection> acquire(Bootstrap b) {
    return Mono.create(sink -> {
        // ...其他省略代码

        SocketAddress remoteAddress = bootstrap.config().remoteAddress();
        PoolKey holder = new PoolKey(remoteAddress, handler != null ? handler.hashCode() : -1);

        // 每个远程地址都可以配置一个PoolFactory，如果没配置则使用默认的PoolFactory
        PoolFactory poolFactory = poolFactoryPerRemoteHost.getOrDefault(remoteAddress, defaultPoolFactory);
        InstrumentedPool<PooledConnection> pool = channelPools.computeIfAbsent(holder, poolKey -> {
            if (log.isDebugEnabled()) {
                log.debug("Creating a new client pool [{}] for [{}]", poolFactory, remoteAddress);
            }

            // newPool是一个连接分配器，实际上就是一个连接池
            InstrumentedPool<PooledConnection> newPool =
                    new PooledConnectionAllocator(bootstrap, poolFactory, opsFactory).pool;

            if (poolFactory.metricsEnabled || BootstrapHandlers.findMetricsSupport(bootstrap) != null) {
                PooledConnectionProviderMetrics.registerMetrics(name,
                        poolKey.hashCode() + "",
                        Metrics.formatSocketAddress(remoteAddress),
                        newPool.metrics());
            }
            return newPool;
        });

        //
        disposableAcquire(new DisposableAcquire(sink, pool, obs, opsFactory, poolFactory.pendingAcquireTimeout, false));

    });

}

static void disposableAcquire(DisposableAcquire disposableAcquire) {
    // accquire一个连接，如果无可用了解则创建，则调用
    Mono<PooledRef<PooledConnection>> mono =
            disposableAcquire.pool.acquire(Duration.ofMillis(disposableAcquire.pendingAcquireTimeout));
    mono.subscribe(disposableAcquire);
}

Publisher<PooledConnection> connectChannel() {
    return Mono.create(sink -> {
        Bootstrap b = bootstrap.clone();
        PooledConnectionInitializer initializer = new PooledConnectionInitializer(sink);
        b.handler(initializer);
        // 创建连接
        ChannelFuture f = b.connect();
        if (f.isDone()) {
            initializer.operationComplete(f);
        } else {
            f.addListener(initializer);
        }
    });
}

从代码里面可以看出，ConnectionProvider对每一个远程地址(即下游服的某一个服务器)都缓存了一个连接分配器(ConnectionAllocator)，而这个ConnectionAllocator才是真正的连接池，是Project Reactor项目内部实现的一个连接池，就不从源码角度分析，简单来说，就是请求方获取连接的时候，如果池子里面有空闲连接，则直接用现成连接，如果没有的话，则调用PoolFactory创建新的链接。

总结一下：

网关内部维持了一个缓存映射，缓存着下游每一个服务地址(ip:port)对应的连接分配器(ConnectionAllocator)，而ConnectionAllocator是一个连接池，内部会保存复用已经生成的连接。

当网关转发请求时确认下游目标服务的地址，即可直接从对应的连接池中取出连接复用。

相关文章