Spring cloud gateway 如何在路由時進行負載均衡
- 2022 年 6 月 9 日
- 筆記
- Spring Cloud
本文為部落客原創，轉載請註明出處：
在網關模組的配置文件中配置路由：
spring:
cloud:
gateway:
routes:
- id: user
uri: lb://user-server
predicates:
- Path=/api-web/** #前端訪問需加入例如 http:ip:port/api-web
filters:
- StripPrefix=1 #訪問後端服務過濾掉m 必填否則找不到後端服務也可以在服務加上統一路徑
其中lb表示採用了負載均衡，user-server表示服務名
當後端有多個服務節點時，網關會以負載均衡的方式將請求發送到後端的各個服務節點上，當某個服務節點關閉以後，後續的請求不會發送到該節點上。這個過程會存在一定的時間延遲，比如30秒左右。
@Configuration(
proxyBeanMethods = false
)
@ConditionalOnClass({LoadBalancerClient.class, RibbonAutoConfiguration.class, DispatcherHandler.class})
@AutoConfigureAfter({RibbonAutoConfiguration.class})
@EnableConfigurationProperties({LoadBalancerProperties.class})
public class GatewayLoadBalancerClientAutoConfiguration {
public GatewayLoadBalancerClientAutoConfiguration() {
}
@Bean
@ConditionalOnBean({LoadBalancerClient.class})
@ConditionalOnMissingBean({LoadBalancerClientFilter.class, ReactiveLoadBalancerClientFilter.class})
public LoadBalancerClientFilter loadBalancerClientFilter(LoadBalancerClient client, LoadBalancerProperties properties) {
// 會載入一個負載均衡的 過濾器 ：LoadBalancerClientFilter
return new LoadBalancerClientFilter(client, properties);
}
}
LoadBalancerClientFilter
public class LoadBalancerClientFilter implements GlobalFilter, Ordered {
public static final int LOAD_BALANCER_CLIENT_FILTER_ORDER = 10100;
private static final Log log = LogFactory.getLog(LoadBalancerClientFilter.class);
protected final LoadBalancerClient loadBalancer;
private LoadBalancerProperties properties;
public LoadBalancerClientFilter(LoadBalancerClient loadBalancer, LoadBalancerProperties properties) {
this.loadBalancer = loadBalancer;
this.properties = properties;
}
public int getOrder() {
return 10100;
}
public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
URI url = (URI)exchange.getAttribute(ServerWebExchangeUtils.GATEWAY_REQUEST_URL_ATTR);
String schemePrefix = (String)exchange.getAttribute(ServerWebExchangeUtils.GATEWAY_SCHEME_PREFIX_ATTR);
if (url != null && ("lb".equals(url.getScheme()) || "lb".equals(schemePrefix))) {
ServerWebExchangeUtils.addOriginalRequestUrl(exchange, url);
if (log.isTraceEnabled()) {
log.trace("LoadBalancerClientFilter url before: " + url);
}
ServiceInstance instance = this.choose(exchange);
if (instance == null) {
throw NotFoundException.create(this.properties.isUse404(), "Unable to find instance for " + url.getHost());
} else {
URI uri = exchange.getRequest().getURI();
String overrideScheme = instance.isSecure() ? "https" : "http";
if (schemePrefix != null) {
overrideScheme = url.getScheme();
}
URI requestUrl = this.loadBalancer.reconstructURI(new DelegatingServiceInstance(instance, overrideScheme), uri);
if (log.isTraceEnabled()) {
log.trace("LoadBalancerClientFilter url chosen: " + requestUrl);
}
exchange.getAttributes().put(ServerWebExchangeUtils.GATEWAY_REQUEST_URL_ATTR, requestUrl);
return chain.filter(exchange);
}
} else {
return chain.filter(exchange);
}
}
protected ServiceInstance choose(ServerWebExchange exchange) {
// 該loadBalancer 為ribbon 配置的負載均衡器，會根據指定的規則進行負載均衡，默認是輪詢
return this.loadBalancer.choose(((URI)exchange.getAttribute(ServerWebExchangeUtils.GATEWAY_REQUEST_URL_ATTR)).getHost());
}
}
NettyRoutingFilter 過濾器：
public class NettyRoutingFilter implements GlobalFilter, Ordered {
private static final Log log = LogFactory.getLog(NettyRoutingFilter.class);
private final HttpClient httpClient;
private final ObjectProvider<List<HttpHeadersFilter>> headersFiltersProvider;
private final HttpClientProperties properties;
private volatile List<HttpHeadersFilter> headersFilters;
public NettyRoutingFilter(HttpClient httpClient, ObjectProvider<List<HttpHeadersFilter>> headersFiltersProvider, HttpClientProperties properties) {
this.httpClient = httpClient;
this.headersFiltersProvider = headersFiltersProvider;
this.properties = properties;
}
public List<HttpHeadersFilter> getHeadersFilters() {
if (this.headersFilters == null) {
this.headersFilters = (List)this.headersFiltersProvider.getIfAvailable();
}
return this.headersFilters;
}
public int getOrder() {
return 2147483647;
}
public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
URI requestUrl = (URI)exchange.getRequiredAttribute(ServerWebExchangeUtils.GATEWAY_REQUEST_URL_ATTR);
String scheme = requestUrl.getScheme();
if (!ServerWebExchangeUtils.isAlreadyRouted(exchange) && ("http".equals(scheme) || "https".equals(scheme))) {
ServerWebExchangeUtils.setAlreadyRouted(exchange);
ServerHttpRequest request = exchange.getRequest();
HttpMethod method = HttpMethod.valueOf(request.getMethodValue());
String url = requestUrl.toASCIIString();
HttpHeaders filtered = HttpHeadersFilter.filterRequest(this.getHeadersFilters(), exchange);
DefaultHttpHeaders httpHeaders = new DefaultHttpHeaders();
filtered.forEach(httpHeaders::set);
boolean preserveHost = (Boolean)exchange.getAttributeOrDefault(ServerWebExchangeUtils.PRESERVE_HOST_HEADER_ATTRIBUTE, false);
Route route = (Route)exchange.getAttribute(ServerWebExchangeUtils.GATEWAY_ROUTE_ATTR);
Flux<HttpClientResponse> responseFlux = ((RequestSender)this.getHttpClient(route, exchange).headers((headers) -> {
headers.add(httpHeaders);
headers.remove("Host");
if (preserveHost) {
String host = request.getHeaders().getFirst("Host");
headers.add("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) -> {
exchange.getAttributes().put(ServerWebExchangeUtils.CLIENT_RESPONSE_ATTR, res);
exchange.getAttributes().put(ServerWebExchangeUtils.CLIENT_RESPONSE_CONN_ATTR, connection);
ServerHttpResponse response = exchange.getResponse();
HttpHeaders headers = new HttpHeaders();
res.responseHeaders().forEach((entry) -> {
headers.add((String)entry.getKey(), (String)entry.getValue());
});
String contentTypeValue = headers.getFirst("Content-Type");
if (StringUtils.hasLength(contentTypeValue)) {
exchange.getAttributes().put("original_response_content_type", contentTypeValue);
}
this.setResponseStatus(res, response);
HttpHeaders filteredResponseHeaders = HttpHeadersFilter.filter(this.getHeadersFilters(), headers, exchange, Type.RESPONSE);
if (!filteredResponseHeaders.containsKey("Transfer-Encoding") && filteredResponseHeaders.containsKey("Content-Length")) {
response.getHeaders().remove("Transfer-Encoding");
}
exchange.getAttributes().put(ServerWebExchangeUtils.CLIENT_RESPONSE_HEADER_NAMES, filteredResponseHeaders.keySet());
response.getHeaders().putAll(filteredResponseHeaders);
return Mono.just(res);
});
Duration responseTimeout = this.getResponseTimeout(route);
if (responseTimeout != null) {
responseFlux = responseFlux.timeout(responseTimeout, Mono.error(new TimeoutException("Response took longer than timeout: " + responseTimeout))).onErrorMap(TimeoutException.class, (th) -> {
return new ResponseStatusException(HttpStatus.GATEWAY_TIMEOUT, th.getMessage(), th);
});
}
return responseFlux.then(chain.filter(exchange));
} else {
return chain.filter(exchange);
}
}
protected ByteBuf getByteBuf(DataBuffer dataBuffer) {
if (dataBuffer instanceof NettyDataBuffer) {
NettyDataBuffer buffer = (NettyDataBuffer)dataBuffer;
return buffer.getNativeBuffer();
} else if (dataBuffer instanceof DefaultDataBuffer) {
DefaultDataBuffer buffer = (DefaultDataBuffer)dataBuffer;
return Unpooled.wrappedBuffer(buffer.getNativeBuffer());
} else {
throw new IllegalArgumentException("Unable to handle DataBuffer of type " + dataBuffer.getClass());
}
}
private void setResponseStatus(HttpClientResponse clientResponse, ServerHttpResponse response) {
HttpStatus status = HttpStatus.resolve(clientResponse.status().code());
if (status != null) {
response.setStatusCode(status);
} else {
while(true) {
if (!(response instanceof ServerHttpResponseDecorator)) {
if (!(response instanceof AbstractServerHttpResponse)) {
throw new IllegalStateException("Unable to set status code " + clientResponse.status().code() + " on response of type " + response.getClass().getName());
}
((AbstractServerHttpResponse)response).setStatusCodeValue(clientResponse.status().code());
break;
}
response = ((ServerHttpResponseDecorator)response).getDelegate();
}
}
}
protected HttpClient getHttpClient(Route route, ServerWebExchange exchange) {
Object connectTimeoutAttr = route.getMetadata().get("connect-timeout");
if (connectTimeoutAttr != null) {
Integer connectTimeout = getInteger(connectTimeoutAttr);
return this.httpClient.tcpConfiguration((tcpClient) -> {
return tcpClient.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, connectTimeout);
});
} else {
return this.httpClient;
}
}
static Integer getInteger(Object connectTimeoutAttr) {
Integer connectTimeout;
if (connectTimeoutAttr instanceof Integer) {
connectTimeout = (Integer)connectTimeoutAttr;
} else {
connectTimeout = Integer.parseInt(connectTimeoutAttr.toString());
}
return connectTimeout;
}
private Duration getResponseTimeout(Route route) {
Object responseTimeoutAttr = route.getMetadata().get("response-timeout");
Long responseTimeout = null;
if (responseTimeoutAttr != null) {
if (responseTimeoutAttr instanceof Number) {
responseTimeout = ((Number)responseTimeoutAttr).longValue();
} else {
responseTimeout = Long.valueOf(responseTimeoutAttr.toString());
}
}
return responseTimeout != null ? Duration.ofMillis(responseTimeout) : this.properties.getResponseTimeout();
}
}
