RestTemplate使用不當引發的問題及解決

背景

系統: SpringBoot開發的Web應用;

ORM: JPA(Hibernate)

接口功能簡述: 根據實體類ID到數據庫中查詢實體信息,然後使用RestTemplate調用外部系統接口獲取數據。

問題現象

瀏覽器頁面有時報504 GateWay Timeout錯誤,刷新多次後,則總是timeout

數據庫連接池報連接耗盡異常

調用外部系統時有時報502 Bad GateWay錯誤

分析過程

為便於描述將本系統稱為A,外部系統稱為B。

這三個問題環環相扣,導火索是第3個問題,然後導致第2個問題,最後導致出現第3個問題;

原因簡述: 第3個問題是由於Nginx負載下沒有掛系統B,導致本系統在請求外部系統時報502錯誤,而A沒有正確處理異常,導致http請求無法正常關閉,而springboot默認打開openSessionInView, 隻有調用A的請求關閉時才會關閉數據庫連接,而此時調用A的請求沒有關閉,導致數據庫連接沒有關閉。

這裡主要分析第1個問題:為什麼請求A的連接出現504 Timeout.

AbstractConnPool

通過日志看到A在調用B時出現阻塞,直到timeout,打印出線程堆棧查看:

線程阻塞在AbstractConnPool類getPoolEntryBlocking方法中

private E getPoolEntryBlocking(
            final T route, final Object state,
            final long timeout, final TimeUnit timeUnit,
            final Future<E> future) throws IOException, InterruptedException, TimeoutException {
        Date deadline = null;
        if (timeout > 0) {
            deadline = new Date (System.currentTimeMillis() + timeUnit.toMillis(timeout));
        }
        this.lock.lock();
        try {
           //根據route獲取route對應的連接池
            final RouteSpecificPool<T, C, E> pool = getPool(route);
            E entry;
            for (;;) {
                Asserts.check(!this.isShutDown, "Connection pool shut down");
                for (;;) {
                   //獲取可用的連接
                    entry = pool.getFree(state);
                    if (entry == null) {
                        break;
                    }
                    // 判斷連接是否過期,如過期則關閉並從可用連接集合中刪除
                    if (entry.isExpired(System.currentTimeMillis())) {
                        entry.close();
                    }
                    if (entry.isClosed()) {
                        this.available.remove(entry);
                        pool.free(entry, false);
                    } else {
                        break;
                    }
                }
               // 如果從連接池中獲取到可用連接,更新可用連接和待釋放連接集合
                if (entry != null) {
                    this.available.remove(entry);
                    this.leased.add(entry);
                    onReuse(entry);
                    return entry;
                }
                // 如果沒有可用連接,則創建新連接
                final int maxPerRoute = getMax(route);
                // 創建新連接之前,檢查是否超過每個route連接池大小,如果超過,則刪除可用連接集合相應數量的連接(從總的可用連接集合和每個route的可用連接集合中刪除)
                final int excess = Math.max(0, pool.getAllocatedCount() + 1 - maxPerRoute);
                if (excess > 0) {
                    for (int i = 0; i < excess; i++) {
                        final E lastUsed = pool.getLastUsed();
                        if (lastUsed == null) {
                            break;
                        }
                        lastUsed.close();
                        this.available.remove(lastUsed);
                        pool.remove(lastUsed);
                    }
                }
                if (pool.getAllocatedCount() < maxPerRoute) {
                   //比較總的可用連接數量與總的可用連接集合大小,釋放多餘的連接資源
                    final int totalUsed = this.leased.size();
                    final int freeCapacity = Math.max(this.maxTotal - totalUsed, 0);
                    if (freeCapacity > 0) {
                        final int totalAvailable = this.available.size();
                        if (totalAvailable > freeCapacity - 1) {
                            if (!this.available.isEmpty()) {
                                final E lastUsed = this.available.removeLast();
                                lastUsed.close();
                                final RouteSpecificPool<T, C, E> otherpool = getPool(lastUsed.getRoute());
                                otherpool.remove(lastUsed);
                            }
                        }
                       // 真正創建連接的地方
                        final C conn = this.connFactory.create(route);
                        entry = pool.add(conn);
                        this.leased.add(entry);
                        return entry;
                    }
                }
               //如果已經超過瞭每個route的連接池大小,則加入隊列等待有可用連接時被喚醒或直到某個終止時間
                boolean success = false;
                try {
                    if (future.isCancelled()) {
                        throw new InterruptedException("Operation interrupted");
                    }
                    pool.queue(future);
                    this.pending.add(future);
                    if (deadline != null) {
                        success = this.condition.awaitUntil(deadline);
                    } else {
                        this.condition.await();
                        success = true;
                    }
                    if (future.isCancelled()) {
                        throw new InterruptedException("Operation interrupted");
                    }
                } finally {
                    //如果到瞭終止時間或有被喚醒時,則出隊,加入下次循環
                    pool.unqueue(future);
                    this.pending.remove(future);
                }
                // 處理異常喚醒和超時情況
                if (!success && (deadline != null && deadline.getTime() <= System.currentTimeMillis())) {
                    break;
                }
            }
            throw new TimeoutException("Timeout waiting for connection");
        } finally {
            this.lock.unlock();
        }
    }

getPoolEntryBlocking方法用於獲取連接,主要有三步:(1).檢查可用連接集合中是否有可重復使用的連接,如果有則獲取連接,返回. (2)創建新連接,註意同時需要檢查可用連接集合(分為每個route的和全局的)是否有多餘的連接資源,如果有,則需要釋放。(3)加入隊列等待;

從線程堆棧可以看出,第1個問題是由於走到瞭第3步。開始時是有時會報504異常,刷新多次後會一直報504異常,經過跟蹤調試發現前幾次會成功獲取到連接,而連接池滿後,後面的請求會阻塞。正常情況下當前面的連接釋放到連接池後,後面的請求會得到連接資源繼續執行,可現實是後面的連接一直處於等待狀態,猜想可能是由於連接一直未釋放導致。

我們來看一下連接在什麼時候會釋放。

RestTemplate

由於在調外部系統B時,使用的是RestTemplate的getForObject方法,從此入手跟蹤調試看一看。

@Override
	public <T> T getForObject(String url, Class<T> responseType, Object... uriVariables) throws RestClientException {
		RequestCallback requestCallback = acceptHeaderRequestCallback(responseType);
		HttpMessageConverterExtractor<T> responseExtractor =
				new HttpMessageConverterExtractor<T>(responseType, getMessageConverters(), logger);
		return execute(url, HttpMethod.GET, requestCallback, responseExtractor, uriVariables);
	}
	@Override
	public <T> T getForObject(String url, Class<T> responseType, Map<String, ?> uriVariables) throws RestClientException {
		RequestCallback requestCallback = acceptHeaderRequestCallback(responseType);
		HttpMessageConverterExtractor<T> responseExtractor =
				new HttpMessageConverterExtractor<T>(responseType, getMessageConverters(), logger);
		return execute(url, HttpMethod.GET, requestCallback, responseExtractor, uriVariables);
	}
	@Override
	public <T> T getForObject(URI url, Class<T> responseType) throws RestClientException {
		RequestCallback requestCallback = acceptHeaderRequestCallback(responseType);
		HttpMessageConverterExtractor<T> responseExtractor =
				new HttpMessageConverterExtractor<T>(responseType, getMessageConverters(), logger);
		return execute(url, HttpMethod.GET, requestCallback, responseExtractor);
	}

getForObject都調用瞭execute方法(其實RestTemplate的其它http請求方法調用的也是execute方法)

@Override
	public <T> T execute(String url, HttpMethod method, RequestCallback requestCallback,
			ResponseExtractor<T> responseExtractor, Object... uriVariables) throws RestClientException {
		URI expanded = getUriTemplateHandler().expand(url, uriVariables);
		return doExecute(expanded, method, requestCallback, responseExtractor);
	}
	@Override
	public <T> T execute(String url, HttpMethod method, RequestCallback requestCallback,
			ResponseExtractor<T> responseExtractor, Map<String, ?> uriVariables) throws RestClientException {
		URI expanded = getUriTemplateHandler().expand(url, uriVariables);
		return doExecute(expanded, method, requestCallback, responseExtractor);
	}
	@Override
	public <T> T execute(URI url, HttpMethod method, RequestCallback requestCallback,
			ResponseExtractor<T> responseExtractor) throws RestClientException {
		return doExecute(url, method, requestCallback, responseExtractor);
	}

所有execute方法都調用瞭同一個doExecute方法

protected <T> T doExecute(URI url, HttpMethod method, RequestCallback requestCallback,
			ResponseExtractor<T> responseExtractor) throws RestClientException {
		Assert.notNull(url, "'url' must not be null");
		Assert.notNull(method, "'method' must not be null");
		ClientHttpResponse response = null;
		try {
			ClientHttpRequest request = createRequest(url, method);
			if (requestCallback != null) {
				requestCallback.doWithRequest(request);
			}
			response = request.execute();
			handleResponse(url, method, response);
			if (responseExtractor != null) {
				return responseExtractor.extractData(response);
			}
			else {
				return null;
			}
		}
		catch (IOException ex) {
			String resource = url.toString();
			String query = url.getRawQuery();
			resource = (query != null ? resource.substring(0, resource.indexOf('?')) : resource);
			throw new ResourceAccessException("I/O error on " + method.name() +
					" request for \"" + resource + "\": " + ex.getMessage(), ex);
		}
		finally {
			if (response != null) {
				response.close();
			}
		}
	}

InterceptingClientHttpRequest

進入到request.execute()方法中,對應抽象類org.springframework.http.client.AbstractClientHttpRequest的execute方法

 @Override
 public final ClientHttpResponse execute() throws IOException {
  assertNotExecuted();
  ClientHttpResponse result = executeInternal(this.headers);
  this.executed = true;
  return result;
 }

調用內部方法executeInternal,executeInternal方法是一個抽象方法,由子類實現(restTemplate內部的http調用實現方式有多種)。進入executeInternal方法,到達抽象類 org.springframework.http.client.AbstractBufferingClientHttpRequest中

 protected ClientHttpResponse executeInternal(HttpHeaders headers) throws IOException {
  byte[] bytes = this.bufferedOutput.toByteArray();
  if (headers.getContentLength() < 0) {
   headers.setContentLength(bytes.length);
  }
  ClientHttpResponse result = executeInternal(headers, bytes);
  this.bufferedOutput = null;
  return result;
 }

緩充請求body數據,調用內部方法executeInternal

ClientHttpResponse result = executeInternal(headers, bytes);

executeInternal方法中調用另一個executeInternal方法,它也是一個抽象方法

進入executeInternal方法,此方法由org.springframework.http.client.AbstractBufferingClientHttpRequest的子類org.springframework.http.client.InterceptingClientHttpRequest實現

 protected final ClientHttpResponse executeInternal(HttpHeaders headers, byte[] bufferedOutput) throws IOException {
  InterceptingRequestExecution requestExecution = new InterceptingRequestExecution();
  return requestExecution.execute(this, bufferedOutput);
 }

實例化瞭一個帶攔截器的請求執行對象InterceptingRequestExecution

  public ClientHttpResponse execute(HttpRequest request, final byte[] body) throws IOException {
              // 如果有攔截器,則執行攔截器並返回結果
   if (this.iterator.hasNext()) {
    ClientHttpRequestInterceptor nextInterceptor = this.iterator.next();
    return nextInterceptor.intercept(request, body, this);
   }
   else {
               // 如果沒有攔截器,則通過requestFactory創建request對象並執行
    ClientHttpRequest delegate = requestFactory.createRequest(request.getURI(), request.getMethod());
    for (Map.Entry<String, List<String>> entry : request.getHeaders().entrySet()) {
     List<String> values = entry.getValue();
     for (String value : values) {
      delegate.getHeaders().add(entry.getKey(), value);
     }
    }
    if (body.length > 0) {
     if (delegate instanceof StreamingHttpOutputMessage) {
      StreamingHttpOutputMessage streamingOutputMessage = (StreamingHttpOutputMessage) delegate;
      streamingOutputMessage.setBody(new StreamingHttpOutputMessage.Body() {
       @Override
       public void writeTo(final OutputStream outputStream) throws IOException {
        StreamUtils.copy(body, outputStream);
       }
      });
      }
     else {
      StreamUtils.copy(body, delegate.getBody());
     }
    }
    return delegate.execute();
   }
  }

InterceptingClientHttpRequest的execute方法,先執行攔截器,最後執行真正的請求對象(什麼是真正的請求對象?見後面攔截器的設計部分)。

看一下RestTemplate的配置:

        RestTemplateBuilder builder = new RestTemplateBuilder();
        return builder
                .setConnectTimeout(customConfig.getRest().getConnectTimeOut())
                .setReadTimeout(customConfig.getRest().getReadTimeout())
                .interceptors(restTemplateLogInterceptor)
                .errorHandler(new ThrowErrorHandler())
                .build();
    }

可以看到配置瞭連接超時,讀超時,攔截器,和錯誤處理器。

看一下攔截器的實現:

    public ClientHttpResponse intercept(HttpRequest httpRequest, byte[] bytes, ClientHttpRequestExecution clientHttpRequestExecution) throws IOException {
        // 打印訪問前日志
        ClientHttpResponse execute = clientHttpRequestExecution.execute(httpRequest, bytes);
        if (如果返回碼不是200) {
            // 拋出自定義運行時異常
        }
        // 打印訪問後日志
        return execute;
    }

可以看到當返回碼不是200時,拋出異常。還記得RestTemplate中的doExecute方法吧,此處如果拋出異常,雖然會執行doExecute方法中的finally代碼,但由於返回的response為null(其實是有response的),沒有關閉response,所以這裡不能拋出異常,如果確實想拋出異常,可以在錯誤處理器errorHandler中拋出,這樣確保response能正常返回和關閉。

RestTemplate源碼部分解析

如何決定使用哪一個底層http框架

知道瞭原因,我們再來看一下RestTemplate在什麼時候決定使用什麼http框架。其實在通過RestTemplateBuilder實例化RestTemplate對象時就決定瞭。

看一下RestTemplateBuilder的build方法

 public RestTemplate build() {
  return build(RestTemplate.class);
 }
 public <T extends RestTemplate> T build(Class<T> restTemplateClass) {
  return configure(BeanUtils.instantiate(restTemplateClass));
 }

可以看到在實例化RestTemplate對象之後,進行配置。可以指定requestFactory,也可以自動探測

 public <T extends RestTemplate> T configure(T restTemplate) {
               // 配置requestFactory
  configureRequestFactory(restTemplate);
        .....省略其它無關代碼
 }
 private void configureRequestFactory(RestTemplate restTemplate) {
  ClientHttpRequestFactory requestFactory = null;
  if (this.requestFactory != null) {
   requestFactory = this.requestFactory;
  }
  else if (this.detectRequestFactory) {
   requestFactory = detectRequestFactory();
  }
  if (requestFactory != null) {
   ClientHttpRequestFactory unwrappedRequestFactory = unwrapRequestFactoryIfNecessary(
     requestFactory);
   for (RequestFactoryCustomizer customizer : this.requestFactoryCustomizers) {
    customizer.customize(unwrappedRequestFactory);
   }
   restTemplate.setRequestFactory(requestFactory);
  }
 }

看一下detectRequestFactory方法

 private ClientHttpRequestFactory detectRequestFactory() {
  for (Map.Entry<String, String> candidate : REQUEST_FACTORY_CANDIDATES
    .entrySet()) {
   ClassLoader classLoader = getClass().getClassLoader();
   if (ClassUtils.isPresent(candidate.getKey(), classLoader)) {
    Class<?> factoryClass = ClassUtils.resolveClassName(candidate.getValue(),
      classLoader);
    ClientHttpRequestFactory requestFactory = (ClientHttpRequestFactory) BeanUtils
      .instantiate(factoryClass);
    initializeIfNecessary(requestFactory);
    return requestFactory;
   }
  }
  return new SimpleClientHttpRequestFactory();
 }

循環REQUEST_FACTORY_CANDIDATES集合,檢查classpath類路徑中是否存在相應的jar包,如果存在,則創建相應框架的封裝類對象。如果都不存在,則返回使用JDK方式實現的RequestFactory對象。

看一下REQUEST_FACTORY_CANDIDATES集合

 private static final Map<String, String> REQUEST_FACTORY_CANDIDATES;
 static {
  Map<String, String> candidates = new LinkedHashMap<String, String>();
  candidates.put("org.apache.http.client.HttpClient",
    "org.springframework.http.client.HttpComponentsClientHttpRequestFactory");
  candidates.put("okhttp3.OkHttpClient",
    "org.springframework.http.client.OkHttp3ClientHttpRequestFactory");
  candidates.put("com.squareup.okhttp.OkHttpClient",
    "org.springframework.http.client.OkHttpClientHttpRequestFactory");
  candidates.put("io.netty.channel.EventLoopGroup",
    "org.springframework.http.client.Netty4ClientHttpRequestFactory");
  REQUEST_FACTORY_CANDIDATES = Collections.unmodifiableMap(candidates);
 }

可以看到共有四種Http調用實現方式,在配置RestTemplate時可指定,並在類路徑中提供相應的實現jar包。

Request攔截器的設計

再看一下InterceptingRequestExecution類的execute方法。

  public ClientHttpResponse execute(HttpRequest request, final byte[] body) throws IOException {
        // 如果有攔截器,則執行攔截器並返回結果
   if (this.iterator.hasNext()) {
   ClientHttpRequestInterceptor nextInterceptor = this.iterator.next();
   return nextInterceptor.intercept(request, body, this);
  }
  else {
         // 如果沒有攔截器,則通過requestFactory創建request對象並執行
      ClientHttpRequest delegate = requestFactory.createRequest(request.getURI(), request.getMethod());
   for (Map.Entry<String, List<String>> entry : request.getHeaders().entrySet()) {
       List<String> values = entry.getValue();
    for (String value : values) {
     delegate.getHeaders().add(entry.getKey(), value);
    }
   }
   if (body.length > 0) {
    if (delegate instanceof StreamingHttpOutputMessage) {
     StreamingHttpOutputMessage streamingOutputMessage = (StreamingHttpOutputMessage) delegate;
     streamingOutputMessage.setBody(new StreamingHttpOutputMessage.Body() {
      @Override
      public void writeTo(final OutputStream outputStream) throws IOException {
       StreamUtils.copy(body, outputStream);
      }
     });
      }
      else {
    StreamUtils.copy(body, delegate.getBody());
      }
   }
   return delegate.execute();
  }
   }

大傢可能會有疑問,傳入的對象已經是request對象瞭,為什麼在沒有攔截器時還要再創建一遍request對象呢?

其實傳入的request對象在有攔截器的時候是InterceptingClientHttpRequest對象,沒有攔截器時,則直接是包裝瞭各個http調用實現框的Request。如HttpComponentsClientHttpRequest、OkHttp3ClientHttpRequest等。當有攔截器時,會執行攔截器,攔截器可以有多個,而這裡 this.iterator.hasNext() 不是一個循環,為什麼呢?秘密在於攔截器的intercept方法。

ClientHttpResponse intercept(HttpRequest request, byte[] body, ClientHttpRequestExecution execution)
      throws IOException;

此方法包含request,body,execution。exection類型為ClientHttpRequestExecution接口,上面的InterceptingRequestExecution便實現瞭此接口,這樣在調用攔截器時,傳入exection對象本身,然後再調一次execute方法,再判斷是否仍有攔截器,如果有,再執行下一個攔截器,將所有攔截器執行完後,再生成真正的request對象,執行http調用。

那如果沒有攔截器呢?

上面已經知道RestTemplate在實例化時會實例化RequestFactory,當發起http請求時,會執行restTemplate的doExecute方法,此方法中會創建Request,而createRequest方法中,首先會獲取RequestFactory

// org.springframework.http.client.support.HttpAccessor
protected ClientHttpRequest createRequest(URI url, HttpMethod method) throws IOException {
   ClientHttpRequest request = getRequestFactory().createRequest(url, method);
   if (logger.isDebugEnabled()) {
      logger.debug("Created " + method.name() + " request for \"" + url + "\"");
   }
   return request;
}
// org.springframework.http.client.support.InterceptingHttpAccessor
public ClientHttpRequestFactory getRequestFactory() {
   ClientHttpRequestFactory delegate = super.getRequestFactory();
   if (!CollectionUtils.isEmpty(getInterceptors())) {
      return new InterceptingClientHttpRequestFactory(delegate, getInterceptors());
   }
   else {
      return delegate;
   }
}

看一下RestTemplate與這兩個類的關系就知道調用關系瞭。

而在獲取到RequestFactory之後,判斷有沒有攔截器,如果有,則創建InterceptingClientHttpRequestFactory對象,而此RequestFactory在createRequest時,會創建InterceptingClientHttpRequest對象,這樣就可以先執行攔截器,最後執行創建真正的Request對象執行http調用。

連接獲取邏輯流程圖

以HttpComponents為底層Http調用實現的邏輯流程圖。

流程圖說明:

  • RestTemplate可以根據配置來實例化對應的RequestFactory,包括apache httpComponents、OkHttp3、Netty等實現。
  • RestTemplate與HttpComponents銜接的類是HttpClient,此類是apache httpComponents提供給用戶使用,執行http調用。HttpClient是創建RequestFactory對象時通過HttpClientBuilder實例化的,在實例化HttpClient對象時,實例化瞭HttpClientConnectionManager和多個ClientExecChain,HttpRequestExecutor、HttpProcessor以及一些策略。
  • 當發起請求時,由requestFactory實例化httpRequest,然後依次執行ClientexecChain,常用的有四種:

RedirectExec:請求跳轉;根據上次響應結果和跳轉策略決定下次跳轉的地址,默認最大執行50次跳轉;

RetryExec:決定出現I/O錯誤的請求是否再次執行

ProtocolExec: 填充必要的http請求header,處理http響應header,更新會話狀態

MainClientExec:請求執行鏈中最後一個節點;從連接池CPool中獲取連接,執行請求調用,並返回請求結果;

  • PoolingHttpClientConnectionManager用於管理連接池,包括連接池初始化,獲取連接,獲取連接,打開連接,釋放連接,關閉連接池等操作。
  • CPool代表連接池,但連接並不保存在CPool中;CPool中維護著三個連接狀態集合:leased(租用的,即待釋放的)/available(可用的)/pending(等待的),用於記錄所有連接的狀態;並且維護著每個Route對應的連接池RouteSpecificPool;
  • RouteSpecificPool是連接真正存放的地方,內部同樣也維護著三個連接狀態集合,但隻記錄屬於本route的連接。
  • HttpComponents將連接按照route劃分連接池,有利於資源隔離,使每個route請求相互不影響;

結束語

在使用框架時,特別是在增強其功能,自定義行為時,要考慮到自定義行為對框架原有流程邏輯的影響,並且最好要熟悉框架相應功能的設計意圖。

在與外部事物交互,包括網絡,磁盤,數據庫等,做到異常情況的處理,保證程序健壯性。

以上為個人經驗,希望能給大傢一個參考,也希望大傢多多支持WalkonNet。

推薦閱讀: