Http 持久连接与 HttpClient 连接池

Posted 业余草

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了Http 持久连接与 HttpClient 连接池相关的知识，希望对你有一定的参考价值。

你知道的越多，不知道的就越多，业余的像一棵小草！

成功路上并不拥挤，因为坚持的人不多。

编辑：业余草

cnblogs.com/kingszelda/p/8988505.html

推荐：https://www.xttblog.com/?p=5231

背景

HTTP 协议是无状态的协议，即每一次请求都是互相独立的。因此它的最初实现是，每一个 http 请求都会打开一个 tcp socket 连接，当交互完毕后会关闭这个连接。

HTTP 协议是全双工的协议，所以建立连接与断开连接是要经过三次握手与四次挥手的。显然在这种设计中，每次发送 Http 请求都会消耗很多的额外资源，即连接的建立与销毁。

于是，HTTP 协议的也进行了发展，通过持久连接的方法来进行 socket 连接复用。

从图中可以看到：

  1. 在串行连接中，每次交互都要打开关闭连接

  2. 在持久连接中，第一次交互会打开连接，交互结束后连接并不关闭，下次交互就省去了建立连接的过程。

持久连接的实现有两种：HTTP/1.0+ 的 keep-alive 与 HTTP/1.1 的持久连接。

HTTP/1.0+ 的 Keep-Alive

从 1996 年开始，很多 HTTP/1.0浏览器与服务器都对协议进行了扩展，那就是“keep-alive”扩展协议。

注意，这个扩展协议是作为 1.0 的补充的“实验型持久连接”出现的。keep-alive 已经不再使用了，最新的 HTTP/1.1 规范中也没有对它进行说明，只是很多应用延续了下来。

使用 HTTP/1.0 的客户端在首部中加上"Connection:Keep-Alive"，请求服务端将一条连接保持在打开状态。服务端如果愿意将这条连接保持在打开状态，就会在响应中包含同样的首部。如果响应中没有包含"Connection:Keep-Alive"首部，则客户端会认为服务端不支持 keep-alive，会在发送完响应报文之后关闭掉当前连接。

通过 keep-alive 补充协议，客户端与服务器之间完成了持久连接，然而仍然存在着一些问题：

  • 在 HTTP/1.0 中 keep-alive 不是标准协议，客户端必须发送Connection:Keep-Alive 来激活 keep-alive 连接。

  • 代理服务器可能无法支持 keep-alive，因为一些代理是"盲中继"，无法理解首部的含义，只是将首部逐跳转发。所以可能造成客户端与服务端都保持了连接，但是代理不接受该连接上的数据。

HTTP/1.1 的持久连接

HTTP/1.1 采取持久连接的方式替代了 Keep-Alive。

HTTP/1.1 的连接默认情况下都是持久连接。如果要显式关闭，需要在报文中加上Connection:Close 首部。即在 HTTP/1.1 中，所有的连接都进行了复用。

然而如同 Keep-Alive 一样，空闲的持久连接也可以随时被客户端与服务端关闭。不发送 Connection:Close 不意味着服务器承诺连接永远保持打开。

HttpClient 如何生成持久连接

HttpClient 中使用了连接池来管理持有连接，同一条 TCP 链路上，连接是可以复用的。HttpClient 通过连接池的方式进行连接持久化。

其实“池”技术是一种通用的设计，其设计思想并不复杂：

  1. 当有连接第一次使用的时候建立连接

  2. 结束时对应连接不关闭，归还到池中

  3. 下次同个目的的连接可从池中获取一个可用连接

  4. 定期清理过期连接

所有的连接池都是这个思路，不过我们看 HttpClient 源码主要关注两点：

  • 连接池的具体设计方案，以供以后自定义连接池参考

  • 如何与 HTTP 协议对应上，即理论抽象转为代码的实现

HttpClient 连接池的实现

HttpClient 关于持久连接的处理在下面的代码中可以集中体现，下面从 MainClientExec 摘取了和连接池相关的部分，去掉了其他部分：

public class MainClientExec implements ClientExecChain {

    @Override
    public CloseableHttpResponse execute(
            final HttpRoute route,
            final HttpRequestWrapper request,
            final HttpClientContext context,
            final HttpExecutionAware execAware) throws IOException, HttpException {
     //从连接管理器HttpClientConnectionManager中获取一个连接请求ConnectionRequest
        final ConnectionRequest connRequest = connManager.requestConnection(route, userToken);final HttpClientConnection managedConn;
        final int timeout = config.getConnectionRequestTimeout();
        //从连接请求ConnectionRequest中获取一个被管理的连接HttpClientConnection
        managedConn = connRequest.get(timeout > 0 ? timeout : 0, TimeUnit.MILLISECONDS);
     //将连接管理器HttpClientConnectionManager与被管理的连接HttpClientConnection交给一个ConnectionHolder持有
        final ConnectionHolder connHolder = new ConnectionHolder(this.log, this.connManager, managedConn);
        try {
            HttpResponse response;
            if (!managedConn.isOpen()) {
          //如果当前被管理的连接不是出于打开状态，需要重新建立连接
                establishRoute(proxyAuthState, managedConn, route, request, context);
            }
       //通过连接HttpClientConnection发送请求
            response = requestExecutor.execute(request, managedConn, context);
       //通过连接重用策略判断是否连接可重用 
            if (reuseStrategy.keepAlive(response, context)) {
                //获得连接有效期
                final long duration = keepAliveStrategy.getKeepAliveDuration(response, context);
                //设置连接有效期
                connHolder.setValidFor(duration, TimeUnit.MILLISECONDS);
          //将当前连接标记为可重用状态
                connHolder.markReusable();
            } else {
                connHolder.markNonReusable();
            }
        }
        final HttpEntity entity = response.getEntity();
        if (entity == null || !entity.isStreaming()) {
            //将当前连接释放到池中，供下次调用
            connHolder.releaseConnection();
            return new HttpResponseProxy(response, null);
        } else {
            return new HttpResponseProxy(response, connHolder);
        }
}

这里看到了在 Http 请求过程中对连接的处理是和协议规范是一致的，这里要展开讲一下具体实现。

PoolingHttpClientConnectionManager 是 HttpClient 默认的连接管理器，首先通过 requestConnection() 获得一个连接的请求，注意这里不是连接。

public ConnectionRequest requestConnection(
            final HttpRoute route,
            final Object state) {final Future<CPoolEntry> future = this.pool.lease(route, state, null);
        return new ConnectionRequest() {
            @Override
            public boolean cancel() {
                return future.cancel(true);
            }
            @Override
            public HttpClientConnection get(
                    final long timeout,
                    final TimeUnit tunit) throws InterruptedException, ExecutionException, ConnectionPoolTimeoutException {
                final HttpClientConnection conn = leaseConnection(future, timeout, tunit);
                if (conn.isOpen()) {
                    final HttpHost host;
                    if (route.getProxyHost() != null) {
                        host = route.getProxyHost();
                    } else {
                        host = route.getTargetHost();
                    }
                    final SocketConfig socketConfig = resolveSocketConfig(host);
                    conn.setSocketTimeout(socketConfig.getSoTimeout());
                }
                return conn;
            }
        };
    }

可以看到返回的 ConnectionRequest 对象实际上是一个持有了 Future，CPoolEntry 是被连接池管理的真正连接实例。

从上面的代码我们应该关注的是：

  • Futurefuture = this.pool.lease(route, state, null)如何从连接池 CPool 中获得一个异步的连接Future<CPoolEntry>

  • HttpClientConnection conn = leaseConnection(future, timeout, tunit)如何通过异步连接 Future 获得一个真正的连接 HttpClientConnection

Future

看一下 CPool 是如何释放一个 Future 的，AbstractConnPool 核心代码如下：

private E getPoolEntryBlocking(
            final T route, final Object state,
            final long timeout, final TimeUnit tunit,
            final Future<E> future) throws IOException, InterruptedException, TimeoutException {
     //首先对当前连接池加锁，当前锁是可重入锁ReentrantLockthis.lock.lock();
        try {
        //获得一个当前HttpRoute对应的连接池，对于HttpClient的连接池而言，总池有个大小，每个route对应的连接也是个池，所以是“池中池”
            final RouteSpecificPool<T, C, E> pool = getPool(route);
            E entry;
            for (;;) {
                Asserts.check(!this.isShutDown, "Connection pool shut down");
          //死循环获得连接
                for (;;) {
            //从route对应的池中拿连接，可能是null，也可能是有效连接
                    entry = pool.getFree(state);
            //如果拿到null，就退出循环
                    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对应的连接最大数量是可配置的，如果超过了，就需要通过LRU清理掉一些连接
                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);
                    }
                }
          //当前route池中的连接数，没有达到上线
                if (pool.getAllocatedCount() < maxPerRoute) {
                    final int totalUsed = this.leased.size();
                    final int freeCapacity = Math.max(this.maxTotal - totalUsed, 0);
            //判断连接池是否超过上线，如果超过了，需要通过LRU清理掉一些连接
                    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);
                            }
                        }
              //根据route建立一个连接
                        final C conn = this.connFactory.create(route);
              //将这个连接放入route对应的“小池”中
                        entry = pool.add(conn);
              //将这个连接放入“大池”中
                        this.leased.add(entry);
                        return entry;
                    }
                }
         //到这里证明没有从获得route池中获得有效连接，并且想要自己建立连接时当前route连接池已经到达最大值，即已经有连接在使用，但是对当前线程不可用
                boolean success = false;
                try {
                    if (future.isCancelled()) {
                        throw new InterruptedException("Operation interrupted");
                    }
            //将future放入route池中等待
                    pool.queue(future);
            //将future放入大连接池中等待
                    this.pending.add(future);
            //如果等待到了信号量的通知,success为true
                    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();
        }
    }

上面的代码逻辑有几个重要点：

  • 连接池有个最大连接数，每个 route 对应一个小连接池，也有个最大连接数

  • 不论是大连接池还是小连接池，当超过数量的时候，都要通过 LRU 释放一些连接

  • 如果拿到了可用连接，则返回给上层使用

  • 如果没有拿到可用连接，HttpClient 会判断当前 route 连接池是否已经超过了最大数量，没有到上限就会新建一个连接，并放入池中

  • 如果到达了上限，就排队等待，等到了信号量，就重新获得一次，等待不到就抛超时异常

  • 通过线程池获取连接要通过 ReetrantLock 加锁，保证线程安全

到这里为止，程序已经拿到了一个可用的 CPoolEntry 实例，或者抛异常终止了程序。

HttpClientConnection

protected HttpClientConnection leaseConnection(
            final Future<CPoolEntry> future,
            final long timeout,
            final TimeUnit tunit) throws InterruptedException, ExecutionException, ConnectionPoolTimeoutException {
        final CPoolEntry entry;
        try {
       //从异步操作Future<CPoolEntry>中获得CPoolEntry
            entry = future.get(timeout, tunit);
            if (entry == null || future.isCancelled()) {
                throw new InterruptedException();
            }
            Asserts.check(entry.getConnection() != null, "Pool entry with no connection");
            if (this.log.isDebugEnabled()) {
                this.log.debug("Connection leased: " + format(entry) + formatStats(entry.getRoute()));
            }
       //获得一个CPoolEntry的代理对象，对其操作都是使用同一个底层的HttpClientConnection
            return CPoolProxy.newProxy(entry);
        } catch (final TimeoutException ex) {
            throw new ConnectionPoolTimeoutException("Timeout waiting for connection from pool");
        }
    }

HttpClient 如何复用持久连接？

在上一章中，我们看到了 HttpClient 通过连接池来获得连接，当需要使用连接的时候从池中获得。

对应着第三章的问题：

  1. 当有连接第一次使用的时候建立连接

  2. 结束时对应连接不关闭，归还到池中

  3. 下次同个目的的连接可从池中获取一个可用连接

  4. 定期清理过期连接

我们在第四章中看到了 HttpClient 是如何处理 1、3 的问题的，那么第 2 个问题是怎么处理的呢？

即 HttpClient 如何判断一个连接在使用完毕后是要关闭，还是要放入池中供他人复用？再看一下 MainClientExec 的代码

//发送Http连接
response = requestExecutor.execute(request, managedConn, context);
//根据重用策略判断当前连接是否要复用
if (reuseStrategy.keepAlive(response, context)) {
    //需要复用的连接，获取连接超时时间，以response中的timeout为准
    final long duration = keepAliveStrategy.getKeepAliveDuration(response, context);
    if (this.log.isDebugEnabled()) {
        final String s;
               //timeout的是毫秒数，如果没有设置则为-1，即没有超时时间
        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();
}

可以看到，当使用连接发生过请求之后，有连接重试策略来决定该连接是否要重用，如果要重用就会在结束后交给 HttpClientConnectionManager 放入池中。

那么连接复用策略的逻辑是怎么样的呢？

public class DefaultClientConnectionReuseStrategy extends DefaultConnectionReuseStrategy {

    public static final DefaultClientConnectionReuseStrategy INSTANCE = new DefaultClientConnectionReuseStrategy();

    @Override
    public boolean keepAlive(final HttpResponse response, final HttpContext context) {
     //从上下文中拿到request
        final HttpRequest request = (HttpRequest) context.getAttribute(HttpCoreContext.HTTP_REQUEST);
        if (request != null) {
       //获得Connection的Header
            final Header[] connHeaders = request.getHeaders(HttpHeaders.CONNECTION);
            if (connHeaders.length != 0) {
                final TokenIterator ti = new BasicTokenIterator(new BasicHeaderIterator(connHeaders, null));
                while (ti.hasNext()) {
                    final String token = ti.nextToken();
            //如果包含Connection:Close首部，则代表请求不打算保持连接，会忽略response的意愿，该头部这是HTTP/1.1的规范
                    if (HTTP.CONN_CLOSE.equalsIgnoreCase(token)) {
                        return false;
                    }
                }
            }
        }
     //使用父类的的复用策略
        return super.keepAlive(response, context);
    }

}

看一下父类的复用策略

if (canResponseHaveBody(request, response)) {
                final Header[] clhs = response.getHeaders(HTTP.CONTENT_LEN);
                //如果reponse的Content-Length没有正确设置，则不复用连接
          //因为对于持久化连接，两次传输之间不需要重新建立连接，则需要根据Content-Length确认内容属于哪次请求，以正确处理“粘包”现象
                //所以，没有正确设置Content-Length的response连接不能复用
                if (clhs.length == 1) {
                    final Header clh = clhs[0];
                    try {
                        final int contentLen = Integer.parseInt(clh.getValue());
                        if (contentLen < 0) {
                            return false;
                        }
                    } catch (final NumberFormatException ex) {
                        return false;
                    }
                } else {
                    return false;
                }
            }
        if (headerIterator.hasNext()) {
            try {
                final TokenIterator ti = new BasicTokenIterator(headerIterator);
                boolean keepalive = false;
                while (ti.hasNext()) {
                    final String token = ti.nextToken();
            //如果response有Connection:Close首部，则明确表示要关闭，则不复用
                    if (HTTP.CONN_CLOSE.equalsIgnoreCase(token)) {
                        return false;
            //如果response有Connection:Keep-Alive首部，则明确表示要持久化，则复用
                    } else if (HTTP.CONN_KEEP_ALIVE.equalsIgnoreCase(token)) {
                        keepalive = true;
                    }
                }
                if (keepalive) {
                    return true;
                }
            } catch (final ParseException px) {
                return false;
            }
        }
     //如果response中没有相关的Connection首部说明，则高于HTTP/1.0版本的都复用连接  
        return !ver.lessEquals(HttpVersion.HTTP_1_0);

总结一下：

  • 如果 request 首部中包含 Connection:Close，不复用

  • 如果 response 中 Content-Length 长度设置不正确，不复用

  • 如果 response 首部包含 Connection:Close，不复用

  • 如果 reponse 首部包含 Connection:Keep-Alive，复用

  • 都没命中的情况下，如果 HTTP 版本高于 1.0 则复用

从代码中可以看到，其实现策略与我们第二、三章协议层的约束是一致的。

HttpClient 如何清理过期连接

在 HttpClient4.4 版本之前，在从连接池中获取重用连接的时候会检查下是否过期，过期则清理。

之后的版本则不同，会有一个单独的线程来扫描连接池中的连接，发现有离最近一次使用超过设置的时间后，就会清理。默认的超时时间是 2 秒钟。

public CloseableHttpClient build() {
            //如果指定了要清理过期连接与空闲连接，才会启动清理线程，默认是不启动的
            if (evictExpiredConnections || evictIdleConnections) {
          //创造一个连接池的清理线程
                final IdleConnectionEvictor connectionEvictor = new IdleConnectionEvictor(cm,
                        maxIdleTime > 0 ? maxIdleTime : 10, maxIdleTimeUnit != null ? maxIdleTimeUnit : TimeUnit.SECONDS,
                        maxIdleTime, maxIdleTimeUnit);
                closeablesCopy.add(new Closeable() {
                    @Override
                    public void close() throws IOException {
                        connectionEvictor.shutdown();
                        try {
                            connectionEvictor.awaitTermination(1L, TimeUnit.SECONDS);
                        } catch (final InterruptedException interrupted) {
                            Thread.currentThread().interrupt();
                        }
                    }

                });
          //执行该清理线程
                connectionEvictor.start();
}

可以看到在 HttpClientBuilder 进行 build 的时候,如果指定了开启清理功能，会创建一个连接池清理线程并运行它。

public IdleConnectionEvictor(
            final HttpClientConnectionManager connectionManager,
            final ThreadFactory threadFactory,
            final long sleepTime, final TimeUnit sleepTimeUnit,
            final long maxIdleTime, final TimeUnit maxIdleTimeUnit) {
        this.connectionManager = Args.notNull(connectionManager, "Connection manager");
        this.threadFactory = threadFactory != null ? threadFactory : new DefaultThreadFactory();
        this.sleepTimeMs = sleepTimeUnit != null ? sleepTimeUnit.toMillis(sleepTime) : sleepTime;
        this.maxIdleTimeMs = maxIdleTimeUnit != null ? maxIdleTimeUnit.toMillis(maxIdleTime) : maxIdleTime;
        this.thread = this.threadFactory.newThread(new Runnable() {
            @Override
            public void run() {
                try {
            //死循环，线程一直执行
                    while (!Thread.currentThread().isInterrupted()) {
              //休息若干秒后执行，默认10秒
                        Thread.sleep(sleepTimeMs);
               //清理过期连接
                        connectionManager.closeExpiredConnections();
               //如果指定了最大空闲时间，则清理空闲连接
                        if (maxIdleTimeMs > 0) {
                            connectionManager.closeIdleConnections(maxIdleTimeMs, TimeUnit.MILLISECONDS);
                        }
                    }
                } catch (final Exception ex) {
                    exception = ex;
                }

            }
        });
    }

总结一下：

  • 只有在 HttpClientBuilder 手动设置后，才会开启清理过期与空闲连接

  • 手动设置后，会启动一个线程死循环执行，每次执行 sleep 一定时间，调用 HttpClientConnectionManager 的清理方法清理过期与空闲连接。

本文总结

  • HTTP 协议通过持久连接的方式，减轻了早期设计中的过多连接问题

  • 持久连接有两种方式：HTTP/1.0+ 的 Keep-Avlive 与 HTTP/1.1 的默认持久连接

  • HttpClient 通过连接池来管理持久连接，连接池分为两个，一个是总连接池，一个是每个 route 对应的连接池

  • HttpClient 通过异步的 Future<CPoolEntry> 来获取一个池化的连接

  • 默认连接重用策略与 HTTP 协议约束一致，根据 response 先判断Connection:Close 则关闭，在判断 Connection:Keep-Alive 则开启，最后版本大于 1.0 则开启

  • 只有在 HttpClientBuilder 中手动开启了清理过期与空闲连接的开关后，才会清理连接池中的连接

  • HttpClient4.4 之后的版本通过一个死循环线程清理过期与空闲连接，该线程每次执行都 sleep 一会，以达到定期执行的效果

上面的研究是基于 HttpClient 源码的个人理解，如果有误，希望大家积极留言讨论。

以上是关于Http 持久连接与 HttpClient 连接池的主要内容，如果未能解决你的问题，请参考以下文章

学习充电Http 持久连接与 HttpClient 连接池

Http 持久连接与 HttpClient 连接池

HttpClient api-连接池

HttpClient连接池耗尽引发雪崩问题

HttpClient连接池

使用 HttpClient 的 HTTP 连接池