HttpClient4.3 连接池参数配置及源码解读

import java.io.IOException;
import java.io.InputStream;
import java.io.InterruptedIOException;
import java.net.UnknownHostException;
import java.nio.charset.CodingErrorAction;
import javax.net.ssl.SSLException;
import org.apache.http.Consts;
import org.apache.http.HttpEntity;
import org.apache.http.HttpEntityEnclosingRequest;
import org.apache.http.HttpHost;
import org.apache.http.HttpRequest;
import org.apache.http.client.HttpRequestRetryHandler;
import org.apache.http.client.config.RequestConfig;
import org.apache.http.client.methods.CloseableHttpResponse;
import org.apache.http.client.methods.HttpGet;
import org.apache.http.client.protocol.HttpClientContext;
import org.apache.http.config.ConnectionConfig;
import org.apache.http.config.MessageConstraints;
import org.apache.http.config.SocketConfig;
import org.apache.http.conn.ConnectTimeoutException;
import org.apache.http.conn.routing.HttpRoute;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.impl.client.DefaultHttpRequestRetryHandler;
import org.apache.http.impl.client.HttpClients;
import org.apache.http.impl.conn.PoolingHttpClientConnectionManager;
import org.apache.http.protocol.HttpContext;public class HttpClientParamTest {
public static void main(String[] args) {
/**
* 创建连接管理器，并设置相关参数
*/
//连接管理器，使用无惨构造
PoolingHttpClientConnectionManager connManager
= new PoolingHttpClientConnectionManager(); /**
* 连接数相关设置
*/
//最大连接数
connManager.setMaxTotal(200);
//默认的每个路由的最大连接数
connManager.setDefaultMaxPerRoute(100);
//设置到某个路由的最大连接数，会覆盖defaultMaxPerRoute
connManager.setMaxPerRoute(new HttpRoute(new HttpHost("somehost", 80)), 150); /**
* socket配置（默认配置 和 某个host的配置）
*/
SocketConfig socketConfig = SocketConfig.custom()
.setTcpNoDelay(true) //是否立即发送数据，设置为true会关闭Socket缓冲，默认为false
.setSoReuseAddress(true) //是否可以在一个进程关闭Socket后，即使它还没有释放端口，其它进程还可以立即重用端口
.setSoTimeout(500) //接收数据的等待超时时间，单位ms
.setSoLinger(60) //关闭Socket时，要么发送完所有数据，要么等待60s后，就关闭连接，此时socket.close()是阻塞的
.setSoKeepAlive(true) //开启监视TCP连接是否有效
.build();
connManager.setDefaultSocketConfig(socketConfig);
connManager.setSocketConfig(new HttpHost("somehost", 80), socketConfig); /**
* HTTP connection相关配置（默认配置 和 某个host的配置）
* 一般不修改HTTP connection相关配置，故不设置
*/
//消息约束
MessageConstraints messageConstraints = MessageConstraints.custom()
.setMaxHeaderCount(200)
.setMaxLineLength(2000)
.build();
//Http connection相关配置
ConnectionConfig connectionConfig = ConnectionConfig.custom()
.setMalformedInputAction(CodingErrorAction.IGNORE)
.setUnmappableInputAction(CodingErrorAction.IGNORE)
.setCharset(Consts.UTF_8)
.setMessageConstraints(messageConstraints)
.build();
//一般不修改HTTP connection相关配置，故不设置
//connManager.setDefaultConnectionConfig(connectionConfig);
//connManager.setConnectionConfig(new HttpHost("somehost", 80), ConnectionConfig.DEFAULT); /**
* request请求相关配置
*/
RequestConfig defaultRequestConfig = RequestConfig.custom()
.setConnectTimeout(2 * 1000) //连接超时时间
.setSocketTimeout(2 * 1000) //读超时时间（等待数据超时时间）
.setConnectionRequestTimeout(500) //从池中获取连接超时时间
.setStaleConnectionCheckEnabled(true)//检查是否为陈旧的连接，默认为true，类似testOnBorrow
.build(); /**
* 重试处理
* 默认是重试3次
*/
//禁用重试(参数：retryCount、requestSentRetryEnabled)
HttpRequestRetryHandler requestRetryHandler = new DefaultHttpRequestRetryHandler(0, false);
//自定义重试策略
HttpRequestRetryHandler myRetryHandler = new HttpRequestRetryHandler() { public boolean retryRequest(IOException exception, int executionCount, HttpContext context) {
//Do not retry if over max retry count
if (executionCount >= 3) {
return false;
}
//Timeout
if (exception instanceof InterruptedIOException) {
return false;
}
//Unknown host
if (exception instanceof UnknownHostException) {
return false;
}
//Connection refused
if (exception instanceof ConnectTimeoutException) {
return false;
}
//SSL handshake exception
if (exception instanceof SSLException) {
return false;
} HttpClientContext clientContext = HttpClientContext.adapt(context);
HttpRequest request = clientContext.getRequest();
boolean idempotent = !(request instanceof HttpEntityEnclosingRequest);
//Retry if the request is considered idempotent
//如果请求类型不是HttpEntityEnclosingRequest，被认为是幂等的，那么就重试
//HttpEntityEnclosingRequest指的是有请求体的request，比HttpRequest多一个Entity属性
//而常用的GET请求是没有请求体的，POST、PUT都是有请求体的
//Rest一般用GET请求获取数据，故幂等，POST用于新增数据，故不幂等
if (idempotent) {
return true;
} return false;
}
}; /**
* 创建httpClient
*/
CloseableHttpClient httpclient = HttpClients.custom()
.setConnectionManager(connManager) //连接管理器
.setProxy(new HttpHost("myproxy", 8080)) //设置代理
.setDefaultRequestConfig(defaultRequestConfig) //默认请求配置
.setRetryHandler(myRetryHandler) //重试策略
.build(); //创建一个Get请求，并重新设置请求参数，覆盖默认
HttpGet httpget = new HttpGet("http://www.somehost.com/");
RequestConfig requestConfig = RequestConfig.copy(defaultRequestConfig)
.setSocketTimeout(5000)
.setConnectTimeout(5000)
.setConnectionRequestTimeout(5000)
.setProxy(new HttpHost("myotherproxy", 8080))
.build();
httpget.setConfig(requestConfig); CloseableHttpResponse response = null;
try {
//执行请求
response = httpclient.execute(httpget); HttpEntity entity = response.getEntity(); // If the response does not enclose an entity, there is no need
// to bother about connection release
if (entity != null) {
InputStream instream = entity.getContent();
try {
instream.read();
// do something useful with the response
}
catch (IOException ex) {
// In case of an IOException the connection will be released
// back to the connection manager automatically
throw ex;
}
finally {
// Closing the input stream will trigger connection release
// 释放连接回到连接池
instream.close();
}
}
}
catch (Exception e) {
e.printStackTrace();
}
finally{
if(response != null){
try {
//关闭连接(如果已经释放连接回连接池，则什么也不做)
response.close();
} catch (IOException e) {
e.printStackTrace();
}
} if(httpclient != null){
try {
//关闭连接管理器，并会关闭其管理的连接
httpclient.close();
} catch (IOException e) {
e.printStackTrace();
}
}
}
}
}

/**
* 创建httpClient
*/
CloseableHttpClient httpclient = HttpClients.custom()
.setConnectionManager(connManager) //连接管理器
.setDefaultRequestConfig(defaultRequestConfig) //默认请求配置
.setRetryHandler(myRetryHandler) //重试策略
.build();

CloseableHttpResponse httpResponse = httpClient.execute(httpUriRequest);

HttpRequestExecutor //发送请求报文，并接收响应信息
MainClientExec(requestExec, connManager, ...) //main Executor，负责连接管理相关
ProtocolExec(execChain, httpprocessor) //HTTP协议封装
RetryExec(execChain, retryHandler) //重试策略
RedirectExec(execChain, routePlanner, redirectStrategy) //重定向

// The connection is in or can be brought to a re-usable state.
// 根据response头中的信息判断是否保持连接
if (reuseStrategy.keepAlive(response, context)) {
// Set the idle duration of this connection
// 根据response头中的keep-alive中的timeout属性，得到连接可以保持的时间（ms）
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);
}
//设置连接保持时间，最终是调用 PoolEntry#updateExpiry
connHolder.setValidFor(duration, TimeUnit.MILLISECONDS);
connHolder.markReusable(); //设置连接reuse=true
}
else {
connHolder.markNonReusable();
}

/**
* This example demonstrates the recommended way of using API to make sure
* the underlying connection gets released back to the connection manager.
*/
public class ClientConnectionRelease { public final static void main(String[] args) throws Exception {
CloseableHttpClient httpclient = HttpClients.createDefault();
try {
HttpGet httpget = new HttpGet("http://localhost/"); System.out.println("Executing request " + httpget.getRequestLine());
CloseableHttpResponse response = httpclient.execute(httpget);
try {
System.out.println("----------------------------------------");
System.out.println(response.getStatusLine()); // Get hold of the response entity
HttpEntity entity = response.getEntity(); // If the response does not enclose an entity, there is no need
// to bother about connection release
if (entity != null) {
InputStream instream = entity.getContent();
try {
instream.read();
// do something useful with the response
} catch (IOException ex) {
// In case of an IOException the connection will be released
// back to the connection manager automatically
throw ex;
} finally {
// Closing the input stream will trigger connection release
instream.close();
}
}
} finally {
response.close();
}
} finally {
httpclient.close();
}
}
}

@Override
public int read(final byte[] b, final int off, final int len) throws IOException {
int l = -1;
if (isReadAllowed()) {
try {
l = wrappedStream.read(b, off, len);
checkEOF(l);
} catch (final IOException ex) {
checkAbort();
throw ex;
}
}
return l;
}

protected void checkEOF(final int eof) throws IOException {
if ((wrappedStream != null) && (eof < 0)) {
try {
boolean scws = true; // should close wrapped stream?
if (eofWatcher != null) {
scws = eofWatcher.eofDetected(wrappedStream);
}
if (scws) {
wrappedStream.close();
}
} finally {
wrappedStream = null;
}
}
}

public boolean eofDetected(final InputStream wrapped) throws IOException {
try {
// there may be some cleanup required, such as
// reading trailers after the response body:
wrapped.close();
releaseConnection(); //释放连接 或 关闭连接
} finally {
cleanup();
}
return false;
}

public void abortConnection() {
synchronized (this.managedConn) {
//如果连接已经释放，直接返回
if (this.released) {
return;
}
this.released = true;
try {
//关闭连接
this.managedConn.shutdown();
log.debug("Connection discarded");
} catch (final IOException ex) {
if (this.log.isDebugEnabled()) {
this.log.debug(ex.getMessage(), ex);
}
} finally {
this.manager.releaseConnection(
this.managedConn, null, 0, TimeUnit.MILLISECONDS);
}
}
}

/**AbstractConnPool#getPoolEntryBlocking()*/
for (;;) {
//从availabe链表头迭代查找符合state的entry
entry = pool.getFree(state);
//找不到entry，跳出
if (entry == null) {
break;
}
//如果entry已关闭或已过期，关闭entry，并从routeToPool、httpConnPool的available队列移除
if (entry.isClosed() || entry.isExpired(System.currentTimeMillis())) {
entry.close();
this.available.remove(entry);
pool.free(entry, false);
}
else { //找到可用连接
break;
}
}

/**MainClientExec#execute()*/
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();
}
}
}

/**BHttpConnectionBase#isStale()*/
public boolean isStale() {
if (!isOpen()) {
return true;
}
try {
final int bytesRead = fillInputBuffer(1);
return bytesRead < 0;
} catch (final SocketTimeoutException ex) {
return false;
} catch (final IOException ex) {
return true;
}
}