httpclient源码分析之MainClientExec
2024-08-26 02:20:09
MainClientExec是HTTP请求处理链中最后一个请求执行环节，负责与另一终端的请求/响应交互，也是很重要的类。
源码版本是4.5.2，主要看execute方法，并在里面添加注释。接着详细说下获取连接的过程。
execute方法
@Override
public CloseableHttpResponse execute(
final HttpRoute route,
final HttpRequestWrapper request,
final HttpClientContext context,
final HttpExecutionAware execAware) throws IOException, HttpException {
Args.notNull(route, "HTTP route");
Args.notNull(request, "HTTP request");
Args.notNull(context, "HTTP context");
//Auth相关，这里没关注
AuthState targetAuthState = context.getTargetAuthState();
if (targetAuthState == null) {
targetAuthState = new AuthState();
context.setAttribute(HttpClientContext.TARGET_AUTH_STATE, targetAuthState);
}
AuthState proxyAuthState = context.getProxyAuthState();
if (proxyAuthState == null) {
proxyAuthState = new AuthState();
context.setAttribute(HttpClientContext.PROXY_AUTH_STATE, proxyAuthState);
}
if (request instanceof HttpEntityEnclosingRequest) {
RequestEntityProxy.enhance((HttpEntityEnclosingRequest) request);
}
//userToken后面作为state，用来从连接池中获取连接的时候使用，默认是null。
//如果设置了值，会设置到连接中，再次获取的时候，则优先取status相等的连接
Object userToken = context.getUserToken();
//ConnectionRequest用来获取HttpClientConnection
//为每一个route设置一个连接池，大小可以配置，默认为2
//从route连接池获取一个连接，优先取status等于userToken的。
//这里没有实质的操作，只是创建一个ConnectionRequest，并将获取连接的操作封装在ConnectionRequest中。
final ConnectionRequest connRequest = connManager.requestConnection(route, userToken);
if (execAware != null) {
if (execAware.isAborted()) {
connRequest.cancel();
throw new RequestAbortedException("Request aborted");
} else {
execAware.setCancellable(connRequest);
}
}
final RequestConfig config = context.getRequestConfig();
final HttpClientConnection managedConn;
try {
final int timeout = config.getConnectionRequestTimeout();
//获取连接，这里才执行从连接池中阻塞获取连接的操作，并设置超时时间。
//这里返回的connection，不一定是有效的socket连接，长短连接处理方式不同。
//如果连接没有打开或者不可用，后面会重新建立socket连接。
managedConn = connRequest.get(timeout > 0 ? timeout : 0, TimeUnit.MILLISECONDS);
} catch(final InterruptedException interrupted) {
Thread.currentThread().interrupt();
throw new RequestAbortedException("Request aborted", interrupted);
} catch(final ExecutionException ex) {
Throwable cause = ex.getCause();
if (cause == null) {
cause = ex;
}
throw new RequestAbortedException("Request execution failed", cause);
}
//将连接加入上下文中，暴露连接。
//context就是一个大容器，收藏各种东西，如果觉得有什么资源是需要在别的地方用到的，那就放入context吧。
context.setAttribute(HttpCoreContext.HTTP_CONNECTION, managedConn);
//是否检查连接的有效性。如果检查不可用，就关闭连接。对于关闭的连接，后面会从三次握手开始，重新建立socket连接。
//如果配置检查，就相当于一个悲观锁，每次请求都会消耗最多30ms来检测，影响性能。4.4版本开始就过时了。
if (config.isStaleConnectionCheckEnabled()) {
// validate connection，首先判断连接是否是打开的
if (managedConn.isOpen()) {
this.log.debug("Stale connection check");
//如果是打开的，进一步判断是否可用
if (managedConn.isStale()) {
this.log.debug("Stale connection detected");
//不可用的时候，需要关闭连接，后面再重新建立连接
managedConn.close();
}
}
}
final ConnectionHolder connHolder = new ConnectionHolder(this.log, this.connManager, managedConn);
try {
if (execAware != null) {
execAware.setCancellable(connHolder);
}
HttpResponse response;
for (int execCount = 1;; execCount++) {
//请求是否幂等的，如果不是，则不能retry，抛异常
if (execCount > 1 && !RequestEntityProxy.isRepeatable(request)) {
throw new NonRepeatableRequestException("Cannot retry request " +
"with a non-repeatable request entity.");
}
if (execAware != null && execAware.isAborted()) {
throw new RequestAbortedException("Request aborted");
}
//如果连接没有打开，即连接使用的socket为null，则重新建立连接。
if (!managedConn.isOpen()) {
this.log.debug("Opening connection " + route);
try {
//建立socket连接。
//遍历地址集，成功建立socket连接，就返回，封装在connection中
establishRoute(proxyAuthState, managedConn, route, request, context);
} catch (final TunnelRefusedException ex) {
if (this.log.isDebugEnabled()) {
this.log.debug(ex.getMessage());
}
response = ex.getResponse();
break;
}
}
final int timeout = config.getSocketTimeout();
if (timeout >= 0) {
//设置socketTimeout
managedConn.setSocketTimeout(timeout);
}
if (execAware != null && execAware.isAborted()) {
throw new RequestAbortedException("Request aborted");
}
if (this.log.isDebugEnabled()) {
this.log.debug("Executing request " + request.getRequestLine());
}
if (!request.containsHeader(AUTH.WWW_AUTH_RESP)) {
if (this.log.isDebugEnabled()) {
this.log.debug("Target auth state: " + targetAuthState.getState());
}
this.authenticator.generateAuthResponse(request, targetAuthState, context);
}
if (!request.containsHeader(AUTH.PROXY_AUTH_RESP) && !route.isTunnelled()) {
if (this.log.isDebugEnabled()) {
this.log.debug("Proxy auth state: " + proxyAuthState.getState());
}
this.authenticator.generateAuthResponse(request, proxyAuthState, context);
}
//和服务器具体交互，发送请求头，如果有响应，再接收响应。
response = requestExecutor.execute(request, managedConn, context);
//根据配置的策略，判断是否保持连接，永久还是一段时长
// The connection is in or can be brought to a re-usable state.
if (reuseStrategy.keepAlive(response, context)) {
// Set the idle duration of this connection
final long duration = keepAliveStrategy.getKeepAliveDuration(response, context);
if (this.log.isDebugEnabled()) {
final String s;
if (duration > 0) {
s = "for " + duration + " " + TimeUnit.MILLISECONDS;
} else {
s = "indefinitely";
}
this.log.debug("Connection can be kept alive " + s);
}
connHolder.setValidFor(duration, TimeUnit.MILLISECONDS);
connHolder.markReusable();
} else {
connHolder.markNonReusable();
}
//跳过
if (needAuthentication(
targetAuthState, proxyAuthState, route, response, context)) {
// Make sure the response body is fully consumed, if present
final HttpEntity entity = response.getEntity();
if (connHolder.isReusable()) {
EntityUtils.consume(entity);
} else {
managedConn.close();
if (proxyAuthState.getState() == AuthProtocolState.SUCCESS
&& proxyAuthState.getAuthScheme() != null
&& proxyAuthState.getAuthScheme().isConnectionBased()) {
this.log.debug("Resetting proxy auth state");
proxyAuthState.reset();
}
if (targetAuthState.getState() == AuthProtocolState.SUCCESS
&& targetAuthState.getAuthScheme() != null
&& targetAuthState.getAuthScheme().isConnectionBased()) {
this.log.debug("Resetting target auth state");
targetAuthState.reset();
}
}
// discard previous auth headers
final HttpRequest original = request.getOriginal();
if (!original.containsHeader(AUTH.WWW_AUTH_RESP)) {
request.removeHeaders(AUTH.WWW_AUTH_RESP);
}
if (!original.containsHeader(AUTH.PROXY_AUTH_RESP)) {
request.removeHeaders(AUTH.PROXY_AUTH_RESP);
}
} else {
break;
}
}
if (userToken == null) {
userToken = userTokenHandler.getUserToken(context);
context.setAttribute(HttpClientContext.USER_TOKEN, userToken);
}
if (userToken != null) {
connHolder.setState(userToken);
}
// check for entity, release connection if possible
//判断是否读取了全部的响应，如果是，则释放连接回连接池，
//否则，也要返回连接，以便后面继续从流中读取响应。
final HttpEntity entity = response.getEntity();
if (entity == null || !entity.isStreaming()) {
// connection not needed and (assumed to be) in re-usable state
connHolder.releaseConnection();
return new HttpResponseProxy(response, null);
} else {
return new HttpResponseProxy(response, connHolder);
}
} catch (final ConnectionShutdownException ex) {
final InterruptedIOException ioex = new InterruptedIOException(
"Connection has been shut down");
ioex.initCause(ex);
throw ioex;
} catch (final HttpException ex) {
connHolder.abortConnection();
throw ex;
} catch (final IOException ex) {
connHolder.abortConnection();
throw ex;
} catch (final RuntimeException ex) {
connHolder.abortConnection();
throw ex;
}
}
总结一下关心的大致流程：
- 创建连接请求
- 根据连接请求的参数，从连接池中获取一个连接
- 配置是否需要校验连接可用性。如果检查不可用，就关闭连接。
- 如果连接没有打开，则创建一个底层的socket连接。
- 发送请求头部（如果请求中带有entity，则发送）
- 如果有响应，接收响应（先接收头部，如果有请求主体，则接收）
这里有一点注意一下：
检测连接有效性的时候，报的是SocketTimeOut异常，而真正读响应的时候，报的是Connection reset异常。为什么不一样呢？我还没找到方法验证，但这里很可能是检测的时间很短，只有1ms，首先触发了SocketTimeOut异常，而实际读响应的时候，是不会这么短时间的。
获取连接
接下来详细说说根据ConnectionRequest获取HttpClientConnection。即：
managedConn = connRequest.get(timeout > 0 ? timeout : 0, TimeUnit.MILLISECONDS);
首先看ConnectionRequest为何物：
//org.apache.http.impl.conn.PoolingHttpClientConnectionManager
@Override
public ConnectionRequest requestConnection(
final HttpRoute route,
final Object state) {
Args.notNull(route, "HTTP route");
if (this.log.isDebugEnabled()) {
this.log.debug("Connection request: " + format(route, state) + formatStats(route));
}
//从连接池中获取一个CPoolEntry（Connection的包装类）
final Future<CPoolEntry> future = this.pool.lease(route, state, null);
return new ConnectionRequest() {
@Override
public boolean cancel() {
return future.cancel(true);
}
// ConnectionRequest的get方法。调用leaseConnection方法，并且传入future（CPoolEntry的封装（connection的封装））
@Override
public HttpClientConnection get(
final long timeout,
final TimeUnit tunit) throws InterruptedException, ExecutionException, ConnectionPoolTimeoutException {
return leaseConnection(future, timeout, tunit);
}
};
}
所以，ConnectionRequest的get方法，实际是调用PoolingHttpClientConnectionManager的leaseConnection，返回一个HttpClientConnection。
关于获取connection的更多详细信息，可以参考这篇文章，详细讲述了PoolingHttpClientConnectionManager的获取连接给用户的方法。
补充
关于config.isStaleConnectionCheckEnabled()：
如果设置每次请求检查连接是否可用，会影响性能。4.4版本开始过时，但官方推荐使用org.apache.http.impl.conn.PoolingHttpClientConnectionManager#getValidateAfterInactivity()。详细了解看这篇最后补充的校验连接有效的方法，有一个案例分析。最后返回的HttpResponseProxy带上ConnectionHolder（响应没有一次读完），这篇文章有一个案例了解，查看成功日志的最后几个步骤。RestTemplate读取扩展字段，第二次读取数据。
另一篇关于此段源码的解读，见这里。
最新文章
- 在Myeclipse中提交代码到GitHub中
- js实现图片加载特效（从左到右，百叶窗，从中间到两边）
- java.lang.ClassNotFoundException: com.mysql.jdbc.Driver
- SAM4E单片机之旅——21、DMAC之USART回显
- [CareerCup] 11.7 Tower of People in Circus 马戏团的人塔
- javascript library
- PPTP模式跟L2TP模式有什么不同
- C#读写文件总结
- PHP set_exception_handler 设置异常处理函数
- bootstrap注意事项（三）
- 2 kNN-K-Nearest Neighbors algorithm k邻近算法（一）
- sqlserver 全库查询 带架构
- ext整合highcharts实现饼图
- WPF防止界面卡死并显示加载中效果
- Flutter终将逆袭！1.2版本发布，或将统一江湖
- Ubuntu16.04下安装elasticsearch+kibana实现php客户端的中文分词
- loadrunner java / JAVA_HOME / CLASSPATH / PATH
- vue webpack 打包后css背景图路径问题
- bug4 导入新工程时报 Target runtime com.genuitec.runtime.generic.jee60 is not defined
- How to Pronounce the I in ING