如何進行SpringCloud Gateway 全鏈路實現分析

本篇文章為大家展示了如何進行SpringCloud Gateway 全鏈路實現分析，內容簡明扼要并且容易理解，絕對能使你眼前一亮，通過這篇文章的詳細介紹希望你能有所收獲。

1. 背景

隨著微服務架構的流行，服務按照不同的維度進行拆分，一次請求往往需要涉及到多個服務。而諸多的服務可能分布在了幾千臺服務器，橫跨多個不同的數據中心。為了快速定位和解決故障，應用性能進行分析，全鏈路監控組件就在這樣的問題背景下產生了。最出名的是谷歌公開的論文提到的 Google Dapper。想要在這個上下文中理解分布式系統的行為，就需要監控那些橫跨了不同的應用、不同的服務器之間的關聯動作。

1.1 全鏈路原理

通過業務調用過程中添加并傳遞調用鏈ID，實現應用間生成鏈路數據，最終串聯成一條完整的調用鏈。 其中整個調用過程中每個請求都要透傳TxId、SpanId和pSpanId。

1.2 Spring Cloud Gateway

作為Spring Cloud官方推出的第二代網關框架，Spring cloud gateway是基于Spring 5.0、Spring Boot2.0和Reactor等技術開發的網關，采用了NIO模型進行通信。

1.2.2 Mono與Flux

Mono表示的是包含 0 或者 1 個元素的異步序列，即要么成功發布元素，要么錯誤

Flux和Mono之間可以相互轉換，比如對一個 Flux 序列進行計數操作，得到的結果是一個 Mono對象，或者把兩個 Mono 序列合并在一起，得到的是一個 Flux 對象。

2. Spring Cloud Gateway不做監控

Spring Cloud Gateway作為入口網關，主要負責服務的路由轉發。如果網關沒有進行監控，則全鏈路會缺失網關節點，直接展示為用戶訪問后續應用；不能有效定位用戶請求慢是網關問題還是后續節點。

3.1.1 Spring Cloud Gateway的請求入口

org.springframework.http.server.reactive.ReactorHttpHandlerAdapter#apply 先將接收到的HttpServerRequest或者最終需要返回的HttpServerResponse包裝轉換為ReactorServerHttpRequest和ReactorServerHttpResponse，再處理請求。

public Mono<Void> apply(HttpServerRequest request, HttpServerResponse response) {

NettyDataBufferFactory bufferFactory = new NettyDataBufferFactory(response.alloc());

ServerHttpRequest adaptedRequest;

ServerHttpResponse adaptedResponse;

try {

adaptedRequest = new ReactorServerHttpRequest(request, bufferFactory);

adaptedResponse = new ReactorServerHttpResponse(response, bufferFactory);

} catch (URISyntaxException ex) {

logger.error("Invalid URL ">

3.1.2 構造網關上下文

org.springframework.web.server.adapter.HttpWebHandlerAdapter#handle

• createExchange()構造網關上下文ServerWebExchange

• getDelegate()通過委托的方式獲取一系列需要處理的WebHandler

public Mono<Void> handle(ServerHttpRequest request, ServerHttpResponse response) {

ServerWebExchange exchange = createExchange(request, response);

return getDelegate().handle(exchange)

.onErrorResume(ex -> handleFailure(request, response, ex))

.then(Mono.defer(response::setComplete));

}

 

protected ServerWebExchange createExchange(ServerHttpRequest request, ServerHttpResponse response) {

return new DefaultServerWebExchange(request, response, this.sessionManager,

getCodecConfigurer(), getLocaleContextResolver(), this.applicationContext);

}

3.1.3 進入Filter鏈

org.springframework.cloud.gateway.handler.FilteringWebHandler#handle 獲得 GatewayFilter 數組，并根據獲得的 GatewayFilter 數組創建DefaultGatewayFilterChain，過濾處理請求。

public Mono<Void> handle(ServerWebExchange exchange) {

Route route = exchange.getRequiredAttribute(GATEWAY_ROUTE_ATTR);

List<GatewayFilter> gatewayFilters = route.getFilters();

List<GatewayFilter> combined = new ArrayList<>(this.globalFilters);

combined.addAll(gatewayFilters);

 

AnnotationAwareOrderComparator.sort(combined);

logger.debug("Sorted gatewayFilterFactories: "+ combined);

return new DefaultGatewayFilterChain(combined).filter(exchange);

}

3.1.4 執行Filter鏈

org.springframework.cloud.gateway.handler.FilteringWebHandler$DefaultGatewayFilterChain#filter 過濾器的鏈式調用

public Mono<Void> filter(ServerWebExchange exchange) {

return Mono.defer(() -> {

if (this.index < filters.size()) {

GatewayFilter filter = filters.get(this.index);

DefaultGatewayFilterChain chain = new DefaultGatewayFilterChain(this, this.index + 1);

return filter.filter(exchange, chain);

} else {

return Mono.empty(); // complete

}

});

}

3.1.5 Gateway Filter適配器

org.springframework.cloud.gateway.handler.FilteringWebHandler$GatewayFilterAdapter#filter GatewayFilterAdapter是GlobalFilter過濾器的包裝類，最終委托Global Filter進行執行。

private static class GatewayFilterAdapter implements GatewayFilter {
	private final GlobalFilter delegate;
	public GatewayFilterAdapter(GlobalFilter delegate) {
		this.delegate = delegate;
	}

	public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
		return this.delegate.filter(exchange, chain);
	}
}

3.1.6 Netty路由網關過濾器

org.springframework.cloud.gateway.filter.NettyRoutingFilter#filter

GlobalFilter實現有很多，此處只分析NettyRoutingFIlter和NettyWriteResponseFilter。而NettyRoutingFilter負責使用 Netty HttpClient 代理對下游的請求。

public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {

// 獲得 requestUrl

URI requestUrl = exchange.getRequiredAttribute(GATEWAY_REQUEST_URL_ATTR);

 

// 判斷是否能夠處理，http或https前綴

String scheme = requestUrl.getScheme();

if (isAlreadyRouted(exchange) || (!"http".equals(scheme) && !"https".equals(scheme))) {

return chain.filter(exchange);

}

// 設置已經路由

setAlreadyRouted(exchange);

 

ServerHttpRequest request = exchange.getRequest();

// 創建Netty Request Method對象

final HttpMethod method = HttpMethod.valueOf(request.getMethod().toString());

final String url = requestUrl.toString();

HttpHeaders filtered = filterRequest(this.headersFilters.getIfAvailable(), exchange);

 

final DefaultHttpHeaders httpHeaders = new DefaultHttpHeaders();

filtered.forEach(httpHeaders::set);

 

String transferEncoding = request.getHeaders().getFirst(HttpHeaders.TRANSFER_ENCODING);

boolean chunkedTransfer = "chunked".equalsIgnoreCase(transferEncoding);

boolean preserveHost = exchange.getAttributeOrDefault(PRESERVE_HOST_HEADER_ATTRIBUTE, false);

// 請求后端服務

return this.httpClient.request(method, url, req -> {

final HttpClientRequest proxyRequest = req.options(NettyPipeline.SendOptions::flushOnEach)

.headers(httpHeaders)

.chunkedTransfer(chunkedTransfer)

.failOnServerError(false)

.failOnClientError(false);

if (preserveHost) {

String host = request.getHeaders().getFirst(HttpHeaders.HOST);

proxyRequest.header(HttpHeaders.HOST, host);

}

return proxyRequest.sendHeaders() //發送請求頭

.send(request.getBody().map(dataBuffer -> // 發送請求Body

((NettyDataBuffer)dataBuffer).getNativeBuffer()));

}).doOnNext(res -> {

ServerHttpResponse response = exchange.getResponse();

// put headers and status so filters can modify the response

HttpHeaders headers = new HttpHeaders();

res.responseHeaders().forEach(entry -> headers.add(entry.getKey(), entry.getValue()));

exchange.getAttributes().put("original_response_content_type", headers.getContentType());

HttpHeaders filteredResponseHeaders = HttpHeadersFilter.filter(

this.headersFilters.getIfAvailable(), headers, exchange, Type.RESPONSE);

response.getHeaders().putAll(filteredResponseHeaders);

 

HttpStatus status = HttpStatus.resolve(res.status().code());

if (status != null) {

response.setStatusCode(status);

} else if (response instanceof AbstractServerHttpResponse) {

// https://jira.spring.io/browse/SPR-16748

((AbstractServerHttpResponse) response).setStatusCodeValue(res.status().code());

} else {

throw new IllegalStateException("Unable to set status code on response: ">

3.1.7 Netty 回寫響應網關過濾器

org.springframework.cloud.gateway.filter.NettyWriteResponseFilter#filter NettyWriteResponseFilter 與NettyRoutingFilter成對出現，負責將代理響應寫回網關客戶端響應。

public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {

// then方法實現After Filter邏輯

return chain.filter(exchange).then(Mono.defer(() -> {

// 獲得Netty Response

HttpClientResponse clientResponse = exchange.getAttribute(CLIENT_RESPONSE_ATTR);

if (clientResponse == null) {

return Mono.empty();

}

log.trace("NettyWriteResponseFilter start");

ServerHttpResponse response = exchange.getResponse();

// 將Netty Response回寫給客戶端

NettyDataBufferFactory factory = (NettyDataBufferFactory) response.bufferFactory();

//TODO: what if it's not netty

final Flux<NettyDataBuffer> body = clientResponse.receive()

.retain() // ByteBufFlux => ByteBufFlux

.map(factory::wrap); // ByteBufFlux => Flux<NettyDataBuffer> 

MediaType contentType = response.getHeaders().getContentType();

return (isStreamingMediaType(contentType) ?

response.writeAndFlushWith(body.map(Flux::just)) : 

response.writeWith(body));

}));

}

4. Spring Cloud Gateway進行監控

當最終將網關也進行監控，可以全局的看到應用請求流轉，網關的請求業務分流，各應用的調用負載情況。

上述內容就是如何進行SpringCloud Gateway 全鏈路實現分析，你們學到知識或技能了嗎？如果還想學到更多技能或者豐富自己的知識儲備，歡迎關注億速云行業資訊頻道。