没有HTTP连接池，空谈什么持久连接

Posted 2021-04-06 魔笛手CTO

 

1

为什么需要HTTP连接池

上世纪90年代初，因为即用即走的“请求—响应”模型，HTTP 协议得以广泛流行。

但是简单并不等同于高效，随着 HTTP 的流行，越来越多的开发者开始抱怨 HTTP 的性能问题。

在这种背景下，HTTP 持久连接应运而生。它改进了原来每一条 HTTP 消息都要新建一条 TCP 连接的低效，允许多个 HTTP 消息复用一个 TCP 连接。

但是 HTTP 持久连接作为协议规范，最终还是需要客户端自己实现。对于客户端而言，如果想要充分利用这一特性，必然需要引入连接池。

那么问题来了，如何实现一个 HTTP 连接池呢？

urllib3 作为最受欢迎的 Python 拓展库之一，为我们提供了一个很好的参考。

2

urllib3 如何实现 HTTP 连接池

（一）

从文档入手

事实上，urllib3 包含二十多个文件，约六七千行代码，即使单单聚焦于连接池也很容易让人望而却步，这时候官方文档就是一个很好的切入点。

先来看一下官方文档给出的使用说明：

  
    
    
  

   
     
     
   >>> import urllib3
   
     
     
   >>> http = urllib3.PoolManager()
   
     
     
   >>> r = http.request('GET', 'http://httpbin.org/robots.txt')
   
     
     
   >>> r.status
   
     
     
   200
   
     
     
   >>> r.data
   
     
     
   'User-agent: *
Disallow: /deny
'
  
    
    
  

从上面的示例虽然简单，但是提供了很好的一个切入点——PoolManager。

（二）

PoolManager

有了切入点，我们就来好好研究一下 PoolManager。

先来看一下 PoolManager 的代码，老张已经关键地方加了注释，相信你肯定可以看懂。

  
    
    
  

   
     
     
   class PoolManager(RequestMethods):
   
     
     
    def __init__(self, num_pools=10, headers=None, **connection_pool_kw):
   
     
     
    RequestMethods.__init__(self, headers)
   
     
     
    self.connection_pool_kw = connection_pool_kw
   
     
     
    # pools故名思意，是连接池的池子
   
     
     
    # 此处实例化的RecentlyUsedContainer看起来像是一个LRU，我们稍后再细看RecentlyUsedContainer
   
     
     
    self.pools = RecentlyUsedContainer(num_pools, dispose_func=lambda p: p.close())
   
     
     
    self.pool_classes_by_scheme = pool_classes_by_scheme
   
     
     
    self.key_fn_by_scheme = key_fn_by_scheme.copy()
   
     
     
   

   
     
     
    def connection_from_pool_key(self, pool_key, request_context=None):
   
     
     
    # 如果大池子中没有对应的连接池，则新建一个
   
     
     
    # 此处返回的pool对象才是真正缓存持久连接的连接池
   
     
     
    with self.pools.lock:
   
     
     
    pool = self.pools.get(pool_key)
   
     
     
    if pool:
   
     
     
    return pool
   
     
     
   

   
     
     
    scheme = request_context["scheme"]
   
     
     
    host = request_context["host"]
   
     
     
    port = request_context["port"]
   
     
     
    # 新建的连接池对象根据协议不同，可能是HTTPConnectionPool或者HTTPSConnectionPool
   
     
     
    # 后续说明以HTTPConnectionPool为主
   
     
     
    pool = self._new_pool(scheme, host, port, request_context=request_context)
   
     
     
    self.pools[pool_key] = pool
   
     
     
   

   
     
     
    return pool
   
     
     
   

   
     
     
    def urlopen(self, method, url, redirect=True, **kw):
   
     
     
    u = parse_url(url)
   
     
     
    self._validate_proxy_scheme_url_selection(u.scheme)
   
     
     
    # self.connection_from_host会根据主机、端口已经协议信息，拼接好pool_key，调用上面的self.connection_from_pool_key
   
     
     
    conn = self.connection_from_host(u.host, port=u.port, scheme=u.scheme)
   
     
     
    # 此处省略后续的代码逻辑
   
     
     
           pass
  
    
    
  

可以看到 PoolManager 实际上连接池的池子，这主要是因为一个客户端对服务器往往是多对多的关系，这就需要针对服务器 S1 和服务器 S2 分别建立对应的连接池，并将它们放到大池子中集中管理。

老张顺手整理一下 PoolManager的UML 类图，方便大家理解。

（三）

RecentlyUsedContainer

通过 PoolManager 的代码，我们看到大池子实际是通过 RecentlyUsedContainer 来缓存连接池的，那么 RecentlyUsedContainer 究竟有何神通？

  
    
    
  

   
     
     
   class RecentlyUsedContainer(MutableMapping):
   
     
     
    """
   
     
     
    RecentlyUsedContainer作为urllib3内部实现的LRU（最近最少使用）结构，保证了线程安全，并且可以像dict一样使用，
   
     
     
    """
   
     
     
    ContainerCls = OrderedDict
   
     
     
   

   
     
     
    def __init__(self, maxsize=10, dispose_func=None):
   
     
     
    self._maxsize = maxsize
   
     
     
    self.dispose_func = dispose_func
   
     
     
    # 内部通过OrderedDict来保存数据
   
     
     
    self._container = self.ContainerCls()
   
     
     
    # 加了线程锁，保证线程安全
   
     
     
    self.lock = RLock()
   
     
     
   

   
     
     
    def __getitem__(self, key):
   
     
     
    # 覆写__getitem__，在返回数据之前刷新其状态至最新
   
     
     
    with self.lock:
   
     
     
    item = self._container.pop(key)
   
     
     
    self._container[key] = item
   
     
     
    return item
   
     
     
   

   
     
     
    def __setitem__(self, key, value):
   
     
     
    # 覆写__setitem__，除了保证数据状态保持最新之外，还确保数据量不会超过容器限制
   
     
     
    evicted_value = _Null
   
     
     
    with self.lock:
   
     
     
    evicted_value = self._container.get(key, _Null)
   
     
     
    self._container[key] = value
   
     
     
   

   
     
     
    if len(self._container) > self._maxsize:
   
     
     
    _key, evicted_value = self._container.popitem(last=False)
   
     
     
    # 此处提供了一个钩子，可以在数据被垃圾回收之前做一些清理操作
   
     
     
    if self.dispose_func and evicted_value is not _Null:
   
     
     
    self.dispose_func(evicted_value)
   
     
     
   

  
    
    
  

RecentlyUsedContainer 是 urllib3 自己实现的一套LUR模型。

RecentlyUsedContainer 选择了继承 Python 内置的抽象类 MutableMapping，这样就可以对外可以像 dict 提供数据操作。

有序字典 OrderedDict 的引入使得 RecentlyUsedContainer 可以很轻松的按照“最近最少使用”原则管理内部数据。

RecentlyUsedContainer 的 UML 类图稍微比 PoolManager 复杂了一些。

（四）

HTTPConnectionPool

了解到了 PoolManager 通过 LRU 来管理连接池，那么真实的连接池又是怎么实现的呢？答案就在 HTTPConnectionPool 的代码里面。

  
    
    
  

   
     
     
   class HTTPConnectionPool(ConnectionPool, RequestMethods):
   
     
     
    scheme = "http"
   
     
     
    # 连接池具体缓存的连接对象为HTTPConnection，我们稍后介绍
   
     
     
    ConnectionCls = HTTPConnection
   
     
     
    ResponseCls = HTTPResponse
   
     
     
   

   
     
     
    def __init__(self, host, port=None, strict=False, timeout=Timeout.DEFAULT_TIMEOUT,
   
     
     
    maxsize=1, block=False, headers=None, retries=None, _proxy=None,
   
     
     
    _proxy_headers=None, _proxy_config=None, **conn_kw):
   
     
     
   

   
     
     
    # 注意：此处省略了部分实例化代码
   
     
     
    # pool作为实际的缓存HTTP连接的数据结构，同大池子一样，有一个最大容量
   
     
     
    # self.QueueCls实际是继承自父类ConnectionPool，具体类型为urllib3自定义的队列结构LifoQueue
   
     
     
    # 稍后会详细介绍LifoQueue
   
     
     
    self.pool = self.QueueCls(maxsize)
   
     
     
    # 省略后续的部分实例化代码
   
     
     
    ...
   
     
     
   

   
     
     
    def _get_conn(self, timeout=None):
   
     
     
    """
   
     
     
    获取HTTP连接的内部私有实例方法，实际调用方为self.urlopen
   
     
     
    """
   
     
     
    conn = None
   
     
     
    # 尝试从连接池获取一条连接
   
     
     
    try:
   
     
     
    conn = self.pool.get(block=self.block, timeout=timeout)
   
     
     
   

   
     
     
    except AttributeError: # self.pool is None
   
     
     
    raise ClosedPoolError(self, "Pool is closed.")
   
     
     
   

   
     
     
    except queue.Empty:
   
     
     
    if self.block:
   
     
     
    raise EmptyPoolError(
   
     
     
    self,
   
     
     
    "Pool reached maximum size and no more connections are allowed.",
   
     
     
    )
   
     
     
    pass 
   
     
     
    # 对于获取的连接做检测
   
     
     
    if conn and is_connection_dropped(conn):
   
     
     
    log.debug("Resetting dropped connection: %s", self.host)
   
     
     
    conn.close()
   
     
     
    if getattr(conn, "auto_open", 1) == 0:
   
     
     
    conn = None
   
     
     
    # 如果缓存的连接为有效连接，直接返回，否则新建一条连接
   
     
     
    return conn or self._new_conn()
   
     
     
   

   
     
     
    def urlopen(self, method, url, body=None, headers=None, retries=None, redirect=True,
   
     
     
    assert_same_host=True, timeout=_Default, pool_timeout=None, release_conn=None,
   
     
     
    chunked=False, body_pos=None, **response_kw
   
     
     
    ):
   
     
     
    # 由于代码太长，这里就不贴urlopen的代码了
   
     
     
    # urlopen的实际功能为从连接池获取一个HTTPConnection实例，发送HTTP请求
   
     
     
    # 另外考虑到从缓存获取的持久连接有可能已经被关闭，此时需要有重试机制
   
     
     
    ...
  
    
    
  

HTTPConnectionPool 作为实际缓存持久连接的连接池，在 urllib3 内部起到承上启下的作用，其属性和方法较多，上面也只是截取了部分跟连接池实现较为相关的部分代码。

（五）

LifoQueue

LifoQueue作为替 HTTPConnectionPool 保存连接的数据结构，从名字 LifoQueue 就可以看出是后入先出的一种队列实现。

  
    
    
  

   
     
     
   class LifoQueue(queue.Queue):
   
     
     
    # 实际上除self.queue不是list外，_qsize、_put、_get方法的实现与Python内置的LifoQueue没有区别
   
     
     
    def _init(self, _):
   
     
     
    self.queue = collections.deque()
   
     
     
   

   
     
     
    def _qsize(self, len=len):
   
     
     
    return len(self.queue)
   
     
     
   

   
     
     
    def _put(self, item):
   
     
     
    # Queue的put方法在保证线程安全以及队列未满的前提下会调用_put，将数据放在队列尾
   
     
     
    self.queue.append(item)
   
     
     
   

   
     
     
    def _get(self):
   
     
     
    # Queue的get方法会在保证线程安全的前提下调用_get，从队列尾部弹出数据
   
     
     
    return self.queue.pop()
  
    
    
  

以上是 LifoQueue 的全部代码，看起来挺简单，但是为什么要用 LifoQueue 呢？

原来，HTTP 的持久连接有一个特点：持久连接往往都会在闲置一段时间后被关闭，这就导致越是新鲜的连接，可用性就越高。

LifoQueue 后入先出的特性，使得其非常适合缓存 HTTP 连接。

（六）

HTTPConnection

既然是连接池，必然少不了的就是连接，HTTPConnection 就是 urllib3 的连接实现，负责发起 HTTP 请求、读取 HTTP 响应。HTTPConnection 继承自Python内置库 http.client.HTTPConnection ，仅仅是覆写了部分接口，感兴趣的小伙伴可以查看老张之前的文章《 》。

  
    
    
  

   
     
     
   class HTTPConnection(_HTTPConnection, object):
   
     
     
    pass
  
    
    
  

3

总结

到这里，我们总结一下通过 urllib3 发起一次 HTTP 请求的完整流程：

1. 通过主机、端口等信息，检查大池子 PoolManager 有没有对应的连接池。
2. 如果 PoolManager 有对应的连接池，跳到第4步。
3. 如果 PoolManager 没有的连接池，新建一个连接池 HTTPConnectionPool 出来，并放入大池子。
4. 检查连接池 HTTPConnectionPool 有没有可用的连接。
5. 如果 HTTPConnectionPool 有可用的连接，从连接池取出，跳到第7步。
6. 如果 HTTPConnectionPool 没有可用的连接，新建一个连接HTTPConnection出来。
7. 通过 HTTPConnection 发送请求，中间可能涉及连接的重试机制。
8. 获得响应之后，将连接 HTTPConnection 放回连接池。

下面是 ullib3 关于连接池的一个整体 UML 类图，感兴趣的小伙伴可以放大研究一下。

最后，有感于 HTTP 连接池，老张又想起了那个真理：工程问题，往往需要取舍。

各位小伙伴有什么想法，也欢迎在下方留言。

魔笛手CTO 发起了一个读者讨论 留言