Spring Cloud Netflix Zuul源碼分析之請求處理篇-下

時間 2019-12-13

標籤 spring cloud netflix zuul 源碼分析請求處理欄目 Spring 简体版

原文原文鏈接

微信公衆號：I am CR7
若有問題或建議，請在下方留言;
最近更新：2019-01-03javascript

前言

因篇幅緣由，上一部份內容請看：Spring Cloud Netflix Zuul源碼分析之請求處理篇-上java

PreDecorationFilter

該類的做用就是查找對應的路由信息，獲取後端微服務的地址，保存到請求上下文，提供給路由過濾器使用。正則表達式

簡化版run方法

 1@Override
 2public Object run() {
 3    RequestContext ctx = RequestContext.getCurrentContext();
 4    final String requestURI = this.urlPathHelper.getPathWithinApplication(ctx.getRequest());
 5    // 根據URI從路由規則裏獲取對應的路由
 6    Route route = this.routeLocator.getMatchingRoute(requestURI);
 7    if (route != null) {
 8        // 將獲取到路由信息放入請求上下文
 9    } else{
10        // 找不到路由，則fallback到DispatcherServlet
11    }
12}
複製代碼

查找路由

時序圖

源碼

這裏特意列出簡單URL是如何查找路由的源碼，請看：後端

 1// SimpleRouteLocator
 2protected ZuulRoute getZuulRoute(String adjustedPath) {
 3    if (!matchesIgnoredPatterns(adjustedPath)) {
 4        for (Entry<String, ZuulRoute> entry : getRoutesMap().entrySet()) {
 5            String pattern = entry.getKey();
 6            log.debug("Matching pattern:" + pattern);
 7            // 利用正則表達式進行path匹配，找到對應的路由
 8            if (this.pathMatcher.match(pattern, adjustedPath)) {
 9                return entry.getValue();
10            }
11        }
12    }
13    return null;
14}
複製代碼

斷點圖

通過該過濾器執行後，請求上下文RequestContext內容爲：
安全

SimpleHostRoutingFilter

該過濾器的做用是調用原生httpClient發送請求到後端微服務，解析響應結果，寫入請求上下文，提供給後置過濾器使用。微信

執行

 1@Override
 2public Object run() {
 3    RequestContext context = RequestContext.getCurrentContext();
 4    HttpServletRequest request = context.getRequest();
 5    // 根據請求構建Zuul請求的Headers
 6    MultiValueMap<String, String> headers = this.helper
 7            .buildZuulRequestHeaders(request);
 8    // 根據請求構建Zuul請求的queryParams
 9    MultiValueMap<String, String> params = this.helper
10            .buildZuulRequestQueryParams(request);
11    String verb = getVerb(request);
12    InputStream requestEntity = getRequestBody(request);
13    if (getContentLength(request) < 0) {
14        context.setChunkedRequestBody();
15    }
16    // 根據請求構建Zuul請求的URI
17    String uri = this.helper.buildZuulRequestURI(request);
18    this.helper.addIgnoredHeaders();
19
20    try {
21        // 調用原生httpClient轉發請求
22        CloseableHttpResponse response = forward(this.httpClient, verb, uri, request,
23                headers, params, requestEntity);
24        // 保存響應結果
25        setResponse(response);
26    }
27    catch (Exception ex) {
28        throw new ZuulRuntimeException(handleException(ex));
29    }
30    return null;
31}
複製代碼

轉發請求

 1private CloseableHttpResponse forward(CloseableHttpClient httpclient, String verb,
 2        String uri, HttpServletRequest request, MultiValueMap<String, String> headers,
 3        MultiValueMap<String, String> params, InputStream requestEntity)
 4        throws Exception {
 5    Map<String, Object> info = this.helper.debug(verb, uri, headers, params,
 6            requestEntity);
 7    // routeHost就是在前置過濾器PreDecorationFilter中添加的
 8    URL host = RequestContext.getCurrentContext().getRouteHost();
 9    // 建立HttpHost，指定請求目標地址
10    HttpHost httpHost = getHttpHost(host);
11    uri = StringUtils.cleanPath(MULTIPLE_SLASH_PATTERN.matcher(host.getPath() + uri).replaceAll("/"));
12    long contentLength = getContentLength(request);
13
14    ContentType contentType = null;
15
16    if (request.getContentType() != null) {
17        contentType = ContentType.parse(request.getContentType());
18    }
19
20    InputStreamEntity entity = new InputStreamEntity(requestEntity, contentLength,
21            contentType);
22
23    // 建立HttpRequest
24    HttpRequest httpRequest = buildHttpRequest(verb, uri, entity, headers, params,
25            request);
26    try {
27        log.debug(httpHost.getHostName() + " " + httpHost.getPort() + " "
28                + httpHost.getSchemeName());
29        // 調用原生httpClient發送請求
30        CloseableHttpResponse zuulResponse = forwardRequest(httpclient, httpHost,
31                httpRequest);
32        this.helper.appendDebug(info, zuulResponse.getStatusLine().getStatusCode(),
33                revertHeaders(zuulResponse.getAllHeaders()));
34        return zuulResponse;
35    }
36    finally {
37        // When HttpClient instance is no longer needed,
38        // shut down the connection manager to ensure
39        // immediate deallocation of all system resources
40        // httpclient.getConnectionManager().shutdown();
41    }
42}
43
44private CloseableHttpResponse forwardRequest(CloseableHttpClient httpclient,
45        HttpHost httpHost, HttpRequest httpRequest) throws IOException {
46    return httpclient.execute(httpHost, httpRequest);
47}
複製代碼

保存響應

 1private void setResponse(HttpResponse response) throws IOException {
 2    RequestContext.getCurrentContext().set("zuulResponse", response);
 3    this.helper.setResponse(response.getStatusLine().getStatusCode(),
 4            response.getEntity() == null ? null : response.getEntity().getContent(),
 5            revertHeaders(response.getAllHeaders()));
 6}
 7// ProxyRequestHelper
 8public void setResponse(int status, InputStream entity,
 9        MultiValueMap<String, String> headers) throws IOException {
10    RequestContext context = RequestContext.getCurrentContext();
11    context.setResponseStatusCode(status);
12    if (entity != null) {
13        context.setResponseDataStream(entity);
14    }
15
16    boolean isOriginResponseGzipped = false;
17    for (Entry<String, List<String>> header : headers.entrySet()) {
18        String name = header.getKey();
19        for (String value : header.getValue()) {
20            context.addOriginResponseHeader(name, value);
21
22            if (name.equalsIgnoreCase(HttpHeaders.CONTENT_ENCODING)
23                    && HTTPRequestUtils.getInstance().isGzipped(value)) {
24                isOriginResponseGzipped = true;
25            }
26            if (name.equalsIgnoreCase(HttpHeaders.CONTENT_LENGTH)) {
27                context.setOriginContentLength(value);
28            }
29            if (isIncludedHeader(name)) {
30                context.addZuulResponseHeader(name, value);
31            }
32        }
33    }
34    context.setResponseGZipped(isOriginResponseGzipped);
35}
複製代碼

斷點圖

SendResponseFilter

該過濾器的做用是將請求上下文裏的響應信息寫入到響應內容，返回給請求客戶端。多線程

執行

 1@Override
 2public Object run() {
 3    try {
 4        addResponseHeaders();
 5        writeResponse();
 6    }
 7    catch (Exception ex) {
 8        ReflectionUtils.rethrowRuntimeException(ex);
 9    }
10    return null;
11}
複製代碼

添加響應Header

 1private void addResponseHeaders() {
 2    RequestContext context = RequestContext.getCurrentContext();
 3    HttpServletResponse servletResponse = context.getResponse();
 4    if (this.zuulProperties.isIncludeDebugHeader()) {
 5        @SuppressWarnings("unchecked")
 6        List<String> rd = (List<String>) context.get(ROUTING_DEBUG_KEY);
 7        if (rd != null) {
 8            StringBuilder debugHeader = new StringBuilder();
 9            for (String it : rd) {
10                debugHeader.append("[[[" + it + "]]]");
11            }
12            servletResponse.addHeader(X_ZUUL_DEBUG_HEADER, debugHeader.toString());
13        }
14    }
15    List<Pair<String, String>> zuulResponseHeaders = context.getZuulResponseHeaders();
16    if (zuulResponseHeaders != null) {
17        for (Pair<String, String> it : zuulResponseHeaders) {
18            servletResponse.addHeader(it.first(), it.second());
19        }
20    }
21    if (includeContentLengthHeader(context)) {
22        Long contentLength = context.getOriginContentLength();
23        if(useServlet31) {
24            servletResponse.setContentLengthLong(contentLength);
25        } else {
26            //Try and set some kind of content length if we can safely convert the Long to an int
27            if (isLongSafe(contentLength)) {
28                servletResponse.setContentLength(contentLength.intValue());
29            }
30        }
31    }
32}
複製代碼

寫出響應

 1private void writeResponse() throws Exception {
 2    RequestContext context = RequestContext.getCurrentContext();
 3    // there is no body to send
 4    if (context.getResponseBody() == null
 5            && context.getResponseDataStream() == null) {
 6        return;
 7    }
 8    HttpServletResponse servletResponse = context.getResponse();
 9    if (servletResponse.getCharacterEncoding() == null) { // only set if not set
10        servletResponse.setCharacterEncoding("UTF-8");
11    }
12
13    OutputStream outStream = servletResponse.getOutputStream();
14    InputStream is = null;
15    try {
16        if (context.getResponseBody() != null) {
17            String body = context.getResponseBody();
18            is = new ByteArrayInputStream(
19                            body.getBytes(servletResponse.getCharacterEncoding()));
20        }
21        else {
22            is = context.getResponseDataStream();
23            if (is!=null && context.getResponseGZipped()) {
24                // if origin response is gzipped, and client has not requested gzip,
25                // decompress stream before sending to client
26                // else, stream gzip directly to client
27                if (isGzipRequested(context)) {
28                    servletResponse.setHeader(ZuulHeaders.CONTENT_ENCODING, "gzip");
29                }
30                else {
31                    is = handleGzipStream(is);
32                }
33            }
34        }
35
36        if (is!=null) {
37            writeResponse(is, outStream);
38        }
39    }
40    finally {
41        if (is != null) {
42            try {
43                is.close();
44            }
45            catch (Exception ex) {
46                log.warn("Error while closing upstream input stream", ex);
47            }
48        }
49
50        try {
51            Object zuulResponse = context.get("zuulResponse");
52            if (zuulResponse instanceof Closeable) {
53                ((Closeable) zuulResponse).close();
54            }
55            outStream.flush();
56            // The container will close the stream for us
57        }
58        catch (IOException ex) {
59            log.warn("Error while sending response to client: " + ex.getMessage());
60        }
61    }
62}
63
64private void writeResponse(InputStream zin, OutputStream out) throws Exception {
65    byte[] bytes = buffers.get();
66    int bytesRead = -1;
67    while ((bytesRead = zin.read(bytes)) != -1) {
68        out.write(bytes, 0, bytesRead);
69    }
70}
複製代碼

小結

經過上述分析，簡單URL請求，Zuul作的哪些事情，就一目瞭然了：併發

根據請求URI正則匹配路由配置規則，找到後端微服務的具體地址
調用原生httpClient發送請求到後端微服務
解析後端微服務的響應結果，返回給請求方

不知道你們有沒有注意到，從始至終有一個類一直貫穿整個處理的過程。誰？Filter？不，是RequestContext。ZuulServlet處理過程就是靠它在前置、路由、後置各過濾器間實現信息傳遞，因而可知它的重要性。接下來，咱們就走進RequestContext，揭開這位「信使」的神祕「面紗」。app

RequestContext

既然RequestContext用來傳遞信息，那麼它的正確性就必須得保證。在多線程狀況下，如何作到這一點呢？請往下看：ide

 1public class RequestContext extends ConcurrentHashMap<String, Object> {
 2
 3    private static final Logger LOG = LoggerFactory.getLogger(RequestContext.class);
 4
 5    protected static Class<? extends RequestContext> contextClass = RequestContext.class;
 6
 7    private static RequestContext testContext = null;
 8
 9    protected static final ThreadLocal<? extends RequestContext> threadLocal = new ThreadLocal<RequestContext>() {
10        // 覆蓋ThreadLocal裏initialValue方法，設置ThreadLocal的值
11        @Override
12        protected RequestContext initialValue() {
13            try {
14                return contextClass.newInstance();
15            } catch (Throwable e) {
16                throw new RuntimeException(e);
17            }
18        }
19    };
20}
複製代碼

是否是明白了一些，對，就是這兩大併發神器：ConcurrentHashMap和ThreadLocal。前面提到的幾個過濾器裏，使用RequestContext時，都是以下寫法：

1RequestContext context = RequestContext.getCurrentContext();
2context.set***();
3context.get***();
複製代碼

getCurrentContext()就是入口：

1public static RequestContext getCurrentContext() {
2    if (testContext != null) return testContext;
3
4    RequestContext context = threadLocal.get();
5    return context;
6}
複製代碼

從ThreadLocal裏獲取RequestContext：

 1public T get() {
 2    Thread t = Thread.currentThread();
 3    ThreadLocalMap map = getMap(t);
 4    if (map != null) {
 5        ThreadLocalMap.Entry e = map.getEntry(this);
 6        if (e != null) {
 7            @SuppressWarnings("unchecked")
 8            T result = (T)e.value;
 9            return result;
10        }
11    }
12    return setInitialValue();
13}
14
15private T setInitialValue() {
16    // 覆蓋該方法，設置ThreadLocal的值
17    T value = initialValue();
18    Thread t = Thread.currentThread();
19    ThreadLocalMap map = getMap(t);
20    if (map != null)
21        map.set(this, value);
22    else
23        createMap(t, value);
24    return value;
25}
複製代碼

經過覆蓋ThreadLocal的initialValue方法，首次調用時設置值，後續調用判斷當前線程已經綁定了值，則直接返回。這樣就保證了不一樣的請求都有本身的RequestContext。由於RequestContext繼承自ConcurrentHashMap，是一個線程安全的map，從而保證了併發下的正確性。

補充：這裏只是簡單講解了ThreadLocal和ConcurrentHashMap，不在本文中詳細展開，後續會寫文章去作深刻分析。

總結

到這裏，咱們就講完了一個簡單URL請求在Zuul中整個處理過程。寫做過程當中，筆者一直在思考，如何行文能讓你們更好的理解。雖然修改了不少次，可是仍是以爲不夠完美，只能一邊寫一邊總結一邊改進。但願你們多多留言，給出意見和建議，那筆者真是感激涕零！！！最後，感謝你們的支持，祝新年快樂，祁琛，2019年1月3日。