关于Spring Cloud Gateway与下游服务器的连接分析
Posted JavaCaiy
tags:
篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了关于Spring Cloud Gateway与下游服务器的连接分析相关的知识,希望对你有一定的参考价值。
背景
最近面试了不少同学,有很大一部分简历上会提到网关,我一般都会顺着往下问他们的网关是怎么做的。
基本上都是说直接使用的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));
上面代码的逻辑主要就是
- 获取通信用的httpclient
- 设置headers,method和url
- 执行responseConnection()方法发起连接
- 连接成功之后执行send()方法传入的lambda方法。
- 执行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是一个连接池,内部会保存复用已经生成的连接。
当网关转发请求时确认下游目标服务的地址,即可直接从对应的连接池中取出连接复用。
以上是关于关于Spring Cloud Gateway与下游服务器的连接分析的主要内容,如果未能解决你的问题,请参考以下文章
在 Spring Cloud Gateway 中禁止未经身份验证的请求
Spring Security 配置:Basic Auth + Spring Cloud Gateway
Spring Cloud Gateway 实现Token校验
关于Spring cloud Gateway集成nacos 实现路由到指定微服务的方式总结