1.spring cloud gateway 配置路由
在網(wǎng)關(guān)模塊的配置文件中配置路由：
spring: cloud: gateway: routes: – id: user uri: lb://user-server predicates: – Path=/api-web/** #前端訪問需加入例如 http:ip:port/api-web filters: – StripPrefix=1 #訪問后端服務(wù)過濾掉m 必填否則找不到后端服務(wù)也可以在服務(wù)加上統(tǒng)一路徑
其中l(wèi)b表示采用了負(fù)載均衡，user-server表示服務(wù)名
當(dāng)后端有多個(gè)服務(wù)節(jié)點(diǎn)時(shí)，網(wǎng)關(guān)會(huì)以負(fù)載均衡的方式將請(qǐng)求發(fā)送到后端的各個(gè)服務(wù)節(jié)點(diǎn)上，當(dāng)某個(gè)服務(wù)節(jié)點(diǎn)關(guān)閉以后，后續(xù)的請(qǐng)求不會(huì)發(fā)送到該節(jié)點(diǎn)上。這個(gè)過程會(huì)存在一定的時(shí)間延遲，比如30秒左右。
2.查看 GatewayLoadBalancerClientAutoConfiguration 的配置類
這個(gè)配置類會(huì)加載一個(gè)過濾器，使用這個(gè)過濾器可以實(shí)現(xiàn)負(fù)載均衡
@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) { // 會(huì)加載一個(gè)負(fù)載均衡的 過濾器 ：LoadBalancerClientFilter return new LoadBalancerClientFilter(client, properties); }}
3.查看 LoadBalancerClientFilter過濾器的實(shí)現(xiàn)
查看該過濾器的實(shí)現(xiàn)
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 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 配置的負(fù)載均衡器，會(huì)根據(jù)指定的規(guī)則進(jìn)行負(fù)載均衡，默認(rèn)是輪詢 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 headersFiltersProvider; private final HttpClientProperties properties; private volatile List headersFilters; public NettyRoutingFilter(HttpClient httpClient, ObjectProvider headersFiltersProvider, HttpClientProperties properties) { this.httpClient = httpClient; this.headersFiltersProvider = headersFiltersProvider; this.properties = properties; } public List getHeadersFilters() { if (this.headersFilters == null) { this.headersFilters = (List)this.headersFiltersProvider.getIfAvailable(); } return this.headersFilters; } public int getOrder() { return 2147483647; } public Mono 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 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(); }}
在NettyRoutingFilter中根據(jù)GATEWAY_REQUEST_URL_ATTR屬性讀取requestUrl，然后進(jìn)行相應(yīng)請(qǐng)求。
LoadBalancerClientFilter會(huì)作用在url以lb開頭的路由，然后利用loadBalancer來獲取服務(wù)實(shí)例，構(gòu)造目標(biāo)requestUrl，設(shè)置到GATEWAY_REQUEST_URL_ATTR屬性中，供NettyRoutingFilter使用。
