HTTP Client MultiThreadedHttpConnectionManager线程安全连接管理类源码解析

为了更好的提供文章，我已经将博客迁移到了自建的博客网站上，我将更多的从源码分析的角度入手，为大家带来更多的深度文章，请大家继续关注我~！  博客地址：www.liuinsect.com

 

_______________________________________________________________________________

 

MultiThreadedHttpConnectionManager 是HTTP Client中用来复用连接的连接管理类,可以通过

 

   

   MultiThreadedHttpConnectionManager n =  new MultiThreadedHttpConnectionManager();
   HttpClient client = new HttpClient(n);

 

 

这样的方式去 创建一个Client 实例，

创建后，每当执行

int statusCode = client.executeMethod(postMethod);时 

http client 委托ConnectionManager创建连接，其实是先委托HttpMethodDirector 执行excute方法，

再通过它委托ConnectionManager 创建连接，HttpMethodDirector 中包含了一下host,请求参数等信息。

 

在创建连接时，HttpMethodDirector 中有如下代码：

         

      if ( this.conn == null) {
          this.conn = connectionManager.getConnectionWithTimeout(
              hostConfiguration,
              this.params.getConnectionManagerTimeout()
           );
          ......
       }

 

ConnectionManager 使用了常用的多态的方式将连接的获取交给子类完成。 增强其扩展性。

ConnectionManager 有三个子类：

 
 

对应于：

1. 一次性的连接：

 

 2. 线程池中获取连接：

 

 

 

3. 复用当前SimpleHttpConnectionManager中的一个成员变量，策略是没有则创建，有则覆盖后返回

 

 

 

重点说下MultiThreadedHttpConnectionManager   中连接的获取：

在使用 MultiThreadedHttpConnectionManager  获取连接的时候，MultiThreadedHttpConnectionManager  使用了连接池的概念针对每个
HostConfiguration 做了连接的管理，即 HostConfiguration 作为Key ，连接池（HostConnectionPool）作为value去管理当前host下的所有连接，
HostConfiguration的实例如下： HostConfiguration[host=http://www.taobao.com]

 

HostConnectionPool 中使用链表 管理了 空闲的连接和等待连接的线程队列。

每次获取连接的时候 根据参数（后面会提到）决定是直接从池中获取一个空闲连接,创建一个连接，还是计算出一个等待时间后 将当前线程沉睡这么久。而后再检查。

Http Client 通过协议对应的ProtocolSocketFactory去创建一个socket连接来发送请求和接受响应

 

使用注意事项：

1. MultiThreadedHttpConnectionManager  中有以下两个变量，分别解释：

     a. 每个host最大同时可以获取的连接数， 大于这个数字后， （1,2号线程正在使用连接）3号线程会wait 沉睡住 直到到达时间或者被打断或者1,2号中有人release这个connection，抛出异常。

          注意，如果是HTTP client 来调用接口的话 这个例如（http://www.taobao.com 那他的host是 www.taobao.com） 这个值应该设置大一点 否则很多线程调用这个接口的时候会阻塞住。

     b. 同一时间MultiThreadedHttpConnectionManager  允许的最大连接数，超过这个数字，连接的建立将会阻塞。直到有空闲连接释放。

 

 

 

使用注意事项测试代码：  下划线的两个方法可以调整后观察结果

 

public static void main(String[] sadfasd) throws HttpException, IOException, InterruptedException{
             final String url= "http://www.taobao.com" ;
             final HttpClient client = new HttpClient();
             final MultiThreadedHttpConnectionManager connectionManager = new MultiThreadedHttpConnectionManager();
             connectionManager.setMaxTotalConnections (1);//总的连接数
             connectionManager.setMaxConnectionsPerHost (2);//每个host的最大连接数
            client.setHttpConnectionManager( connectionManager );
            
            Runnable r = new Runnable(){
                   public  void run(){
                         int statusCode=0;
                        PostMethod postMethod = new PostMethod(url);
                         try {
                              statusCode = client.executeMethod(postMethod);
                              System. out.println( "sleep" + statusCode );
                              Thread. sleep(3000);//10s
                              postMethod.releaseConnection();
                        } catch (HttpException e) {
                              e.printStackTrace();
                        } catch (IOException e) {
                              e.printStackTrace();
                        } catch (InterruptedException e) {
                              e.printStackTrace();
                        }
                  };
            };
            Runnable r1 = new Runnable(){
                   public  void run(){
                         int statusCode=0;
                        PostMethod postMethod = new PostMethod(url);
                         try {
                              statusCode = client.executeMethod(postMethod);
                        } catch (HttpException e) {
                              e.printStackTrace();
                        } catch (IOException e) {
                              e.printStackTrace();
                        }
                        System. out.println( statusCode );
                        postMethod.releaseConnection();
                  };
            };
            Runnable r2 = new Runnable(){
                   public  void run(){
                         int statusCode=0;
                        PostMethod postMethod = new PostMethod(url);
                         try {
                              statusCode = client.executeMethod(postMethod);
                        } catch (HttpException e) {
                              e.printStackTrace();
                        } catch (IOException e) {
                              e.printStackTrace();
                        }
                        System. out.println( statusCode );
                        postMethod.releaseConnection();
                  };
            };
            Runnable r3 = new Runnable(){
                   public  void run(){
                         int statusCode=0;
                        PostMethod postMethod = new PostMethod(url);
                         try {
                              statusCode = client.executeMethod(postMethod);
                        } catch (HttpException e) {
                              e.printStackTrace();
                        } catch (IOException e) {
                              e.printStackTrace();
                        }
                        System. out.println( statusCode );
                        postMethod.releaseConnection();
                  };
            };
             new Thread(r).start();
            Thread. sleep(1000);
             new Thread(r1).start();
             new Thread(r2).start();
             new Thread(r3).start();
            
      }

 

 

释放连接：

在我们调用postMethod.releaseConnection()时， 会调用connectionManager的releaseConnection方法。
注意：进入这个方法后会首先同步整个connectionPool（连接池）对象，这意味着，在多连接复用的时候频繁的释放连接，也是会有性能损耗的，同步整个connectionPool后连接的创建都会受影响。
然后开始归还连接，归还的方式很清晰：

1. 将Connection放到基于host的连接池的空闲链表中
    hostPool. freeConnections .add(conn);
2.将Connection放到整个全局的connectionPool的空闲链表中
3. 将Connection从Reference Map中移除（Reference Map 后面单独讲解）
4. 将Connection加入到超时管理中去。
5. 将hostPool（host连接池）里等待队列的头元素拿出来 发送interrupt的信号量。目的是 唤醒等待连接的线程。

 

到目前为止，有两个点可以详细说下
1. Reference Map的作用。
2.  等待连接的线程的处理方式。

 

首先说Reference Map,这个名字是我自己取的。它在MultiThreadedHttpConnectionManager  中的名字叫做：

 

在每次获取连接和释放连接的时候会将”连接“存入和移除。

注意： 这里的”连接“ 已经不是Connection 而是用 WeakReference包装过的Connection。

 

为什么用WeakReference？

这里的概念和ThreadLocal 中用WeakReference 包装ThreadLocalMap中的Key一样。 

目的是为了 在连接丢失时，HTTP client 失去了对“连接”（Connection）的强引用，该连接对象变成了弱引用对象，可以被GC掉。

所以，每次在获取连接的时候 要将连接用WeakReference 包装后放到REFERENCE_TO_CONNECTION_SOURCE 这个Map中,

每次释放连接时，将它从REFERENCE_TO_CONNECTION_SOURCE 中移除,因为这个时候连接的管理由线程池使用强引用管理。

 

 

再说，等待连接的线程的处理方式

先看 获取连接时的代码 和注释  大部分代码被精简了。 所以逻辑不通，看流程即可。

 

  synchronized (connectionPool) {
          while (connection == null) {
              if (hostPool.freeConnections.size() > 0) { 
                    //有线程池中有空闲的连接
                  connection = connectionPool.getFreeConnection(hostConfiguration);
 
              } else if ((hostPool.numConnections < maxHostConnections) && (connectionPool.numConnections < maxTotalConnections)) {
                    //没有空闲连接，但是满足前文的两个条件 可以创建新的连接
                  connection = connectionPool.createConnection(hostConfiguration); 
 
              } else if ((hostPool.numConnections < maxHostConnections) && (connectionPool.freeConnections.size() > 0)) {
 
                    //整个连接数 没有到达最大，并且有空闲连接（其他host池中） 则删除掉其他host中的连接，并且在当前host池子中创建新连接
                  connectionPool.deleteLeastUsedConnection();
                  connection = connectionPool.createConnection(hostConfiguration);
              } else {
                  //以上条件都不满足， 只能将当前线程睡眠
                  try { 
                    waitingThread = new WaitingThread();//创建一个线程包装类
                    waitingThread.hostConnectionPool = hostPool;//指定所属的host连接池
                    waitingThread.thread = Thread.currentThread();//将当前线程赋值           
                    startWait = System.currentTimeMillis ();
                     
                      hostPool.waitingThreads.addLast(waitingThread);//将线程包装类 添加到host连接池的 等待列表中
                      connectionPool.waitingThreads.addLast(waitingThread);//将线程包装类 添加到全局连接池的 等待列表中
                      connectionPool.wait(timeToWait);//沉睡
                  } catch (InterruptedException e) {
                         //被打断是检查 布尔变量interruptedByConnectionPool 确定是 HTTP 释放连接后 主动打断的，还是其他异常原因打断
                         //是自己打断的 catch住异常后什么也不做，重新进入while循环中，尝试获取连接
                           if (!waitingThread.interruptedByConnectionPool) {
                               throw new IllegalThreadStateException("Interrupted while waiting in MultiThreadedHttpConnectionManager");
                           }
                  } finally {
                      if (!waitingThread.interruptedByConnectionPool) {
                          hostPool.waitingThreads.remove(waitingThread);
                          connectionPool.waitingThreads.remove(waitingThread);
                      }
                      if (useTimeout) {
                          endWait = System.currentTimeMillis ();
                          timeToWait -= (endWait - startWait);
                      }
                  }
              }
          }
      }

 

 

释放连接时

调用notifyWaitingThread 方法，结合上面的代码看：

 

 public synchronized void notifyWaitingThread(HostConnectionPool hostPool) {
 
         // find the thread we are going to notify, we want to ensure that each
         // waiting thread is only interrupted once so we will remove it from
         // all wait queues before interrupting it
         WaitingThread waitingThread = null;
         // 取出 等待的线程后发送     interrupt 信号量，
        
         if (hostPool.waitingThreads.size() > 0) {
            
             waitingThread = ( WaitingThread) hostPool.waitingThreads.removeFirst();
             waitingThreads.remove(waitingThread);
         } else if (waitingThreads .size() > 0) {
            
             waitingThread = ( WaitingThread) waitingThreads.removeFirst();
             waitingThread.hostConnectionPool.waitingThreads.remove(waitingThread);
         }
         // 导致 获取连接的那个方法中 捕获异常
         // 注：interrupt 信号量是一定会引起 interruptException的
         // 将interruptedByConnectionPool 设置为true 好标明 是 HTTP client 手动打断的。 这是HTTP client对于等待线程唤醒方式的核心思路
         if (waitingThread != null) {
             waitingThread.interruptedByConnectionPool = true;
             waitingThread.thread.interrupt();
         }
     }

 

 

上面两端代码主要思路就是: 有空连接就直接用，没有则沉睡等待唤醒。

其实用interrupt信号量 会引起interruptException异常，通过catch住异常来处理，是比较粗暴的。

优雅的用 wait and notify的方式 就不需要catch异常，同样能达到唤醒线程效果，而且很优雅。

 

MultiThreadedHttpConnectionManager  中对弱引用的使用

MultiThreadedHttpConnectionManager  类中 还有一个 ReferenceQueueThread类 是用来配合HttpConnectionWithReference（将连接用弱引用包裹后的对象）使用的

 

 

使用的方式是这样：

 

1. 创建连接时，用弱引用包裹住Connection对象放到REFERENCE_TO_CONNECTION_SOURCE  中，目的是防止在连接丢失的时候Map中的这个HttpConnectionWithReference 对象变成弱引用，

     在GC回收时会被回收掉，防止内存泄露。

2. 首先明确的是，JVM会在HttpConnectionWithReference 被回收的时候，将他加入到REFERENCE_QUEUE 中。这是JAVA对于弱引用的规则。

3. 同时，在将HttpConnectionWithReference  放入Map时，启动一个子线程 ReferenceQueueThread  去监听 这个REFERENCE_QUEUE ，只要这个REFERENCE_QUEUE  有值（被GC回收的时候）

     立马被取出来，将线程池可用连接的大小 -1 。

 

MultiThreadedHttpConnectionManager  使用弱引用 确保了

1. connection对象丢失时 内存的及时回收。

2. 搭配队列和子线程确保，连接丢失后线程池中可用连接数的次数可以修改。

 

说到这里，HTTP Client的MultiThreadedHttpConnectionManager  类的绝大部分分方法已经解释完毕了。其中主要是省略掉了，发送和读取HTTP 报文的代码，没有太多技巧，以规则解析出来即可。

 

总结：

 

1. 在单纯的发送请求的场景下，使用MultiThreadedHttpConnectionManager 来代替SimpleHTTPConnectionManger是可行的，并且MultiThreadedHttpConnectionManager 的连接池机制也会提高发送请求的效率，

2. 但是觉得不符合分布式应用间的借口调用，原因很简单，对每个host做了连接池，在一定情况下，这个限制是致命的，直接影响了接口的调用效率。严重影响调用的并发数。所以，在分布式应用的调用中不适合使用MultiThreadedHttpConnectionManager 。

 

MultiThreadedHttpConnectionManager类中几个值得注意的点：

 

1. 连接的管理，特别是使用WeakReference包装Connection对象，然后结合一个子线程和Queque去确保对象被回收时，可以连接数的增加。

2. 对于没有连接可用时，使用使当前线程睡眠的，在释放连接时 使用 interrupt信号量 是等待线程恢复的处理方式

 

 

 

 

评论

4 楼 liuInsect 2013-06-15  

ziwuzu 写道

liuInsect 写道

ziwuzu 写道

为什么不适合分布式应用？有什么效率问题？

在没有一个统一的 远程调用框架的时候 用HTTP Client 是可以的，或者说只能用HTTP Client

但是使用这种连接池的Manager 不合适的原因是 它的连接池 是基于Host的。同一个Host的最大请求量受连接池大小限制。 你说适合分布式环境么？

如果是连接池大小的问题，可以修改连接池的大小。

但是它低连接数的时候几十个，高并发的时候 峰值成百上千个，怎么搞？
是不是都因为设置的这个值阻塞在这里了~？

3 楼 ziwuzu 2013-06-15  

liuInsect 写道

ziwuzu 写道

为什么不适合分布式应用？有什么效率问题？

在没有一个统一的 远程调用框架的时候 用HTTP Client 是可以的，或者说只能用HTTP Client

但是使用这种连接池的Manager 不合适的原因是 它的连接池 是基于Host的。同一个Host的最大请求量受连接池大小限制。 你说适合分布式环境么？

如果是连接池大小的问题，可以修改连接池的大小。

2 楼 liuInsect 2013-06-14  

ziwuzu 写道

为什么不适合分布式应用？有什么效率问题？

在没有一个统一的 远程调用框架的时候 用HTTP Client 是可以的，或者说只能用HTTP Client

但是使用这种连接池的Manager 不合适的原因是 它的连接池 是基于Host的。同一个Host的最大请求量受连接池大小限制。 你说适合分布式环境么？

1 楼 ziwuzu 2013-06-14  

为什么不适合分布式应用？有什么效率问题？