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请求相互不影响；
结束语
在使用框架时，特别是在增强其功能，自定义行为时，要考虑到自定义行为对框架原有流程逻辑的影响，并且最好要熟悉框架相应功能的设计意图。
在与外部事物交互，包括网络，磁盘，数据库等，做到异常情况的处理，保证程序健壮性。
以上为个人经验，希望能给大家一个参考，也希望大家多多支持。
上一篇： JavaScript 沙箱探索
推荐阅读
-
为JS扩展Array.prototype.indexOf引发的问题及解决办法_基础知识
-
Oracle RAC之－－安装过程中碰到的问题及解决方法
-
js中for循环内的匿名函数使用i的问题及解决方案
-
解决dubbo错误ip及ip乱入问题的方法
-
iOS tableView上拉刷新显示下载进度的问题及解决办法
-
微信小程序 swiper 组件遇到的问题及解决方法
-
ios 11和iphone x的相关适配问题及解决方法
-
C3P0连接池+MySQL的配置及wait_timeout问题的解决方法
-
Xcode 8打印log日志的问题小结及解决方法
-
Xcode8下iOS10常见报错闪退，字体适配和编译不过的问题及解决方案