httpclient源码分析之 PoolingHttpClientConnectionManager 获取连接 （转）

 public PoolingHttpClientConnectionManager(
        final HttpClientConnectionOperator httpClientConnectionOperator,
        final HttpConnectionFactory connFactory,
        final long timeToLive, final TimeUnit tunit) {
        super();
        this.configData = new ConfigData();
        //defaultMaxPerRoute默认为2，maxTotal默认为20
        this.pool = new CPool(new InternalConnectionFactory(
                this.configData, connFactory), 2, 20, timeToLive, tunit);
        //validateAfterInactivity 空闲永久连接检查间隔，这个牵扯的还比较多
        //官方推荐使用这个来检查永久链接的可用性，而不推荐每次请求的时候才去检查        
        this.pool.setValidateAfterInactivity(2000);
        this.connectionOperator = Args.notNull(httpClientConnectionOperator, "HttpClientConnectionOperator");
        this.isShutDown = new AtomicBoolean(false);
    }

 @Override
    public ConnectionRequest requestConnection(
            final HttpRoute route,
            final Object state) {
        ......
        
        //从连接池中获取一个CPoolEntry（Connection的包装类）
        final Future future = this.pool.lease(route, state, null);
        return new ConnectionRequest() {
			......

            // 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);
            }
        };
    }


    protected HttpClientConnection leaseConnection(
            final Future future,
            final long timeout,
            final TimeUnit tunit) throws InterruptedException, ExecutionException, ConnectionPoolTimeoutException {
        final CPoolEntry entry;
        try {
            //从future中get
            entry = future.get(timeout, tunit);
            if (entry == null || future.isCancelled()) {
                throw new InterruptedException();
            }
            Asserts.check(entry.getConnection() != null, "Pool entry with no connection");

            //封装返回
            return CPoolProxy.newProxy(entry);
        } catch (final TimeoutException ex) {
            throw new ConnectionPoolTimeoutException("Timeout waiting for connection from pool");
        }
    }

 //org.apache.http.pool.AbstractConnPool
    @Override
    public Future lease(final T route, final Object state, final FutureCallback callback) {
        ......

        return new PoolEntryFuture(this.lock, callback) {
            @Override
            public E getPoolEntry(
                    final long timeout,
                    final TimeUnit tunit)
                        throws InterruptedException, TimeoutException, IOException {
                //阻塞获取CPoolEntry        
                final E entry = getPoolEntryBlocking(route, state, timeout, tunit, this);
                onLease(entry);
                return entry;
            }

        };
    }

    private E getPoolEntryBlocking(
            final T route, final Object state,
            final long timeout, final TimeUnit tunit,
            final PoolEntryFuture future)
                throws IOException, InterruptedException, TimeoutException {
        //设置超时时间点        
        ......

        //串行操作
        this.lock.lock();
        try {
            //每一个route都有一个连接池，这里获取指定route的连接池
            final RouteSpecificPool pool = getPool(route);
            E entry = null;
            //循环取，直到超时
            while (entry == null) {
                Asserts.check(!this.isShutDown, "Connection pool shut down");
                for (;;) {
                    //从连接池中去一个空闲的连接，优先取state相同的。state默认是null
                    entry = pool.getFree(state);
                    //如果没有符合的连接，则调出，创建一个新连接
                    if (entry == null) {
                        break;
                    }
                    //如果连接超时，则关闭
                    if (entry.isExpired(System.currentTimeMillis())) {
                        entry.close();
                    //如果是永久连接，且最近周期内没有检验，则校验连接是否可用。不可用的连接需要关闭    
                    } else if (this.validateAfterInactivity > 0) {
                        if (entry.getUpdated() + this.validateAfterInactivity 

//如果往前validateAfterInactivity ms之内没有校验，则校验entry。校验不通过则关闭并释放，继续从连接池中获取entry。
//如果往前validateAfterInactivity ms之内有过校验，则无需再次校验
 if (entry.getUpdated() + this.validateAfterInactivity 

//org.apache.http.impl.conn.CPool
    protected boolean validate(final CPoolEntry entry) {
        return !entry.getConnection().isStale();
    }

	//org.apache.http.impl.BHttpConnectionBase
	//判断连接是否不可用(go down)
    public boolean isStale() {
	    //没有打开，即socket为空，则不可用
        if (!isOpen()) {
            return true;
        }
        try {
        //socket链路有了，测试链路是否可用
        //这里的测试方法是查看很短的时间内（这里是1ms），是否可以从输入流中读到数据
        //如果测试结果返回-1说明不可用
            final int bytesRead = fillInputBuffer(1);
            return bytesRead 

 @Override
    public int read(final byte[] b, final int off, final int len) throws IOException {
        try {
            final int bytesRead = in.read(b, off, len);
            if (bytesRead == -1) {
                wire.input("end of stream");
            } else if (bytesRead > 0) {
                wire.input(b, off, bytesRead);
            }
            return bytesRead;
        } catch (final IOException ex) {
            wire.input("[read] I/O error: " + ex.getMessage());
            throw ex;
        }
    }