HTTP Client MultiThreadedHttpConnectionManager线程安全连接管理类源码解析
HTTP Client MultiThreadedHttpConnectionManager线程安全连接管理类源码解析
九州暮云 发表于8个月前
HTTP Client MultiThreadedHttpConnectionManager线程安全连接管理类源码解析
  • 发表于 8个月前
  • 阅读 17
  • 收藏 1
  • 点赞 0
  • 评论 0

标题:腾讯云 新注册用户域名抢购1元起>>>   

摘要: 本文主要对MultiThreadedHttpConnectionManager类进行解析

MultiThreadedHttpConnectionManager 解释

MultiThreadedHttpConnectionManager 是HTTP Client中用来复用连接的连接管理类,可以通过这样的方式去 创建一个Client 实例:

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

创建后,每当执行 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,主要是由于它的连接池是基于Host的,同一个Host的最大请求量受连接池大小限制。

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

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

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

使用httpclient的经验:

  1. 创建的MultiThreadedHttpConnectionManager 实例最好是全局的,否则会有多个连接池

  2. HttpClient可以多个实例。

  3. 处理请求的最后,要调用method.releaseConnection();回收连接,释放资源。

参考链接:

Apache Commons Http Client - 2 minute tutorial

共有 人打赏支持
粉丝 46
博文 123
码字总数 62582
×
九州暮云
如果觉得我的文章对您有用,请随意打赏。您的支持将鼓励我继续创作!
* 金额(元)
¥1 ¥5 ¥10 ¥20 其他金额
打赏人
留言
* 支付类型
微信扫码支付
打赏金额:
已支付成功
打赏金额: