如何用python實(shí)現(xiàn)一個(gè)HTTP連接池

更新時(shí)間：2021年01月14日 11:05:33 作者：終末之冬

這篇文章主要介紹了如何用python實(shí)現(xiàn)一個(gè)HTTP連接池的步驟，幫助大家更好的理解和使用python，感興趣的朋友可以了解下

一. 連接池的原理

　　首先, HTTP連接是基于TCP連接的, 與服務(wù)器之間進(jìn)行HTTP通信, 本質(zhì)就是與服務(wù)器之間建立了TCP連接后, 相互收發(fā)基于HTTP協(xié)議的數(shù)據(jù)包. 因此, 如果我們需要頻繁地去請(qǐng)求某個(gè)服務(wù)器的資源, 我們就可以一直維持與個(gè)服務(wù)器的TCP連接不斷開(kāi), 然后在需要請(qǐng)求資源的時(shí)候, 把連接拿出來(lái)用就行了.

　　一個(gè)項(xiàng)目可能需要與服務(wù)器之間同時(shí)保持多個(gè)連接, 比如一個(gè)爬蟲(chóng)項(xiàng)目, 有的線程需要請(qǐng)求服務(wù)器的網(wǎng)頁(yè)資源, 有的線程需要請(qǐng)求服務(wù)器的圖片等資源, 而這些請(qǐng)求都可以建立在同一條TCP連接上.

　　因此, 我們使用一個(gè)管理器來(lái)對(duì)這些連接進(jìn)行管理, 任何程序需要使用這些連接時(shí), 向管理器申請(qǐng)就可以了, 等到用完之后再將連接返回給管理器, 以供其他程序重復(fù)使用, 這個(gè)管理器就是連接池.

二. 實(shí)現(xiàn)代碼

1. HTTPConnectionPool類

　　基于上一章的分析, 連接池應(yīng)該是一個(gè)收納連接的容器, 同時(shí)對(duì)這些連接有管理能力:

class HTTPConnectionPool:

 def __init__(self, host: str, port: int = None, max_size: int = None, idle_timeout: int = None) -> None:
  """
  :param host: pass
  :param port: pass
  :param max_size: 同時(shí)存在的最大連接數(shù), 默認(rèn)None->連接數(shù)無(wú)限,沒(méi)了就創(chuàng)建
  :param idle_timeout: 單個(gè)連接單次最長(zhǎng)空閑時(shí)間,超時(shí)自動(dòng)關(guān)閉,默認(rèn)None->不限時(shí)
  """
  self.host = host
  self.port = port
  self.max_size = max_size
  self.idle_timeout = idle_timeout
  self._lock = threading.Condition()
  self._pool = []
  # 這里的conn_num指的是總連接數(shù),包括其它線程拿出去正在使用的連接
  self.conn_num = 0
  self.is_closed = False

 def acquire(self, blocking: bool = True, timeout: int = None) -> WrapperHTTPConnection:
  ...

 def release(self, conn: WrapperHTTPConnection) -> None:
  ...

　　因此, 我們定義這樣一個(gè)HTTPConnectionPool類, 使用一個(gè)列表來(lái)保存可用的連接. 對(duì)于外部來(lái)說(shuō), 只需要調(diào)用這個(gè)連接池對(duì)象的acquire和release方法就能取得和釋放連接.

2. 線程安全地管理連接

　　對(duì)于線程池內(nèi)部來(lái)說(shuō), 至少需要三個(gè)關(guān)于連接的操作: 從連接池中取得連接, 將連接放回連接池, 以及創(chuàng)建一個(gè)連接:

def _get_connection(self) -> WrapperHTTPConnection:
 # 這個(gè)方法會(huì)把連接從_idle_conn移動(dòng)到_used_conn列表中,并返回這個(gè)連接
 try:
  return self._pool.pop()
 except IndexError:
  raise EmptyPoolError


def _put_connection(self, conn: WrapperHTTPConnection) -> None:
 self._pool.append(conn)


def _create_connection(self) -> WrapperHTTPConnection:
 self.conn_num += 1
 return WrapperHTTPConnection(self, HTTPConnection(self.host, self.port))

　　對(duì)于連接池外部來(lái)說(shuō), 主要有申請(qǐng)連接和釋放連接這兩個(gè)操作, 實(shí)際上這就是個(gè)簡(jiǎn)單的生產(chǎn)者消費(fèi)者模型. 考慮到外部可能是多線程的環(huán)境, 我們使用threading.Condition來(lái)保證線程安全. 關(guān)于Condition的資料可以看這里.

def acquire(self, blocking: bool = True, timeout: int = None) -> WrapperHTTPConnection:
 if self.is_closed:
  raise ConnectionPoolClosed
 with self._lock:
  if self.max_size is None or not self.is_full():
   # 在還能創(chuàng)建新連接的情況下,如果沒(méi)有空閑連接,直接創(chuàng)建一個(gè)就行了
   if self.is_pool_empty():
    self._put_connection(self._create_connection())
  else:
   # 不能創(chuàng)建新連接的情況下,如果設(shè)置了blocking=False,沒(méi)連接就報(bào)錯(cuò)
   # 否則,就基于timeout進(jìn)行阻塞,直到超時(shí)或者有可用連接為止
   if not blocking:
    if self.is_pool_empty():
     raise EmptyPoolError
   elif timeout is None:
    while self.is_pool_empty():
     self._lock.wait()
   elif timeout < 0:
    raise ValueError("'timeout' must be a non-negative number")
   else:
    end_time = time.time() + timeout
    while self.is_pool_empty():
     remaining = end_time - time.time()
     if remaining <= 0:
      raise EmptyPoolError
     self._lock.wait(remaining)
  # 走到這一步了,池子里一定有空閑連接
  return self._get_connection()


def release(self, conn: WrapperHTTPConnection) -> None:
 if self.is_closed:
  # 如果這個(gè)連接是在連接池關(guān)閉后才釋放的,那就不用回連接池了,直接放生
  conn.close()
  return
 # 實(shí)際上,python列表的append操作是線程安全的,可以不加鎖
 # 這里調(diào)用鎖是為了通過(guò)notify方法通知其它正在wait的線程:現(xiàn)在有連接可用了
 with self._lock:
  if not conn.is_available:
   # 如果這個(gè)連接不可用了,就應(yīng)該創(chuàng)建一個(gè)新連接放進(jìn)去,因?yàn)榭赡苓€有其它線程在等著連接用　　　　　　 conn.close()　　　　　　 self.conn_num -= 1
   conn = self._create_connection()
  self._put_connection(conn)
  self._lock.notify()

　　我們首先看看acquire方法, 這個(gè)方法其實(shí)就是在申請(qǐng)到鎖之后調(diào)用內(nèi)部的_get_connection方法獲取連接, 這樣就線程安全了. 需要注意的是, 如果當(dāng)前的條件無(wú)法獲取連接, 就會(huì)調(diào)用條件變量的wait方法, 及時(shí)釋放鎖并阻塞住當(dāng)前線程. 然后, 當(dāng)其它線程作為生產(chǎn)者調(diào)用release方法釋放連接時(shí), 會(huì)觸發(fā)條件變量的notify方法, 從而喚醒一個(gè)阻塞在wait階段的線程, 即消費(fèi)者. 這個(gè)消費(fèi)者再?gòu)某刂腥〕鰟偡呕厝サ木€程, 這樣整個(gè)生產(chǎn)者消費(fèi)者模型就運(yùn)轉(zhuǎn)起來(lái)了.

3. 上下文管理器

　　對(duì)于一個(gè)程序來(lái)說(shuō), 它使用連接池的形式是獲取連接->使用連接->釋放連接. 因此, 我們應(yīng)該通過(guò)with語(yǔ)句來(lái)管理這個(gè)連接, 以免在程序的最后遺漏掉釋放連接這一步驟.

　　基于這個(gè)原因, 我們通過(guò)一個(gè)WrapperHTTPConnection類來(lái)對(duì)HTTPConnection進(jìn)行封裝, 以實(shí)現(xiàn)上下文管理器的功能. HTTPConnection的代碼可以看《用python實(shí)現(xiàn)一個(gè)HTTP客戶端》這篇文章.

class WrapperHTTPConnection:

 def __init__(self, pool: 'HTTPConnectionPool', conn: HTTPConnection) -> None:
  self.pool = pool
  self.conn = conn
  self.response = None
  self.is_available = True

 def __enter__(self) -> 'WrapperHTTPConnection':
  return self

 def __exit__(self, *exit_info: Any) -> None:
  # 如果response沒(méi)讀完并且連接需要復(fù)用,就棄用這個(gè)連接
  if not self.response.will_close and not self.response.is_closed():
   self.close()
  self.pool.release(self)

 def request(self, *args: Any, **kwargs: Any) -> HTTPResponse:
  self.conn.request(*args, **kwargs)
  self.response = self.conn.get_response()
  return self.response

 def close(self) -> None:
  self.conn.close()
  self.is_available = False

　　同樣的, 連接池可能也需要關(guān)閉, 因此我們給連接池也加上上下文管理器的功能:

class HTTPConnectionPool:
 ...

 def close(self) -> None:
  if self.is_closed:
   return
  self.is_closed = True
  pool, self._pool = self._pool, None
  for conn in pool:
   conn.close()

 def __enter__(self) -> 'HTTPConnectionPool':
  return self

 def __exit__(self, *exit_info: Any) -> None:
  self.close()

　　這樣, 我們就可以通過(guò)with語(yǔ)句優(yōu)雅地管理連接池了:

with HTTPConnectionPool(**kwargs) as pool:
 with pool.acquire() as conn:
  res = conn.request('GET', '/')
  ...

4. 定時(shí)清理連接

　　如果一個(gè)連接池的所需連接數(shù)是隨時(shí)間變化的, 那么就會(huì)出現(xiàn)一種情況: 在高峰期, 我們創(chuàng)建了非常多的連接, 然后進(jìn)入低谷期之后, 連接過(guò)剩, 大量的連接處于空閑狀態(tài), 浪費(fèi)資源. 因此, 我們可以設(shè)置一個(gè)定時(shí)任務(wù), 定期清理空閑時(shí)間過(guò)長(zhǎng)的連接, 減少連接池的資源占用.

　　首先, 我們需要為連接對(duì)象添加一個(gè)last_time屬性, 每當(dāng)連接釋放進(jìn)入連接池后, 就修改這個(gè)屬性的值為當(dāng)前時(shí)間, 這樣我們就能明確知道, 連接池內(nèi)的每個(gè)空閑連接空閑了多久:

class WrapperHTTPConnection:
 ...

 def __init__(self, pool: 'HTTPConnectionPool', conn: HTTPConnection) -> None:
  ...
  self.last_time = None


class HTTPConnectionPool:
 ...

 def _put_connection(self, conn: WrapperHTTPConnection) -> None:
  conn.last_time = time.time()
  self._pool.append(conn)

　　然后, 我們通過(guò)threading.Timer來(lái)實(shí)現(xiàn)一個(gè)定時(shí)任務(wù):

def start_clear_conn(self) -> None:
 if self.idle_timeout is None:
  # 如果空閑連接的超時(shí)時(shí)間為無(wú)限,那么就不應(yīng)該清理連接
  return
 self.clear_idle_conn()
 self._clearer = threading.Timer(self.idle_timeout, self.start_clear_conn)
 self._clearer.start()


def stop_clear_conn(self) -> None:
 if self._clearer is not None:
  self._clearer.cancel()

　　threading.Timer只會(huì)執(zhí)行一次定時(shí)任務(wù), 因此, 我們需要在start_clear_conn中不斷地把自己設(shè)置為定時(shí)任務(wù). 這其實(shí)等同于新開(kāi)了一個(gè)線程來(lái)執(zhí)行start_clear_conn方法, 因此并不會(huì)出現(xiàn)遞歸過(guò)深問(wèn)題. 不過(guò)需要注意的是, threading.Timer雖然不會(huì)阻塞當(dāng)前線程, 但是卻會(huì)阻止當(dāng)前線程結(jié)束, 就算把它設(shè)置為守護(hù)線程都不行, 唯一可行的辦法就是調(diào)用stop_clear_conn方法取消這個(gè)定時(shí)任務(wù).

　　最后, 我們定義clear_idle_conn方法來(lái)清理閑置時(shí)間超時(shí)的連接:

def clear_idle_conn(self) -> None:
 if self.is_closed:
  raise ConnectionPoolClosed
 # 這里開(kāi)一個(gè)新線程來(lái)清理空閑連接,避免了阻塞主線程導(dǎo)致的定時(shí)精度出錯(cuò)
 threading.Thread(target=self._clear_idle_conn).start()


def _clear_idle_conn(self) -> None:
 if not self._lock.acquire(timeout=self.idle_timeout):
  # 因?yàn)槭敲扛魋elf.idle_timeout秒檢查一次
  # 如果過(guò)了self.idle_timeout秒還沒(méi)申請(qǐng)到鎖,下一次都開(kāi)始了,本次也就不用繼續(xù)了
  return
 current_time = time.time()
 if self.is_pool_empty():
  pass
 elif current_time - self._pool[-1].last_time >= self.idle_timeout:
  # 這里處理下面的二分法沒(méi)法處理的邊界情況,即所有連接都閑置超時(shí)的情況
  self.conn_num -= len(self._pool)
  self._pool.clear()
 else:
  # 通過(guò)二分法找出從左往右第一個(gè)不超時(shí)的連接的指針
  left, right = 0, len(self._pool) - 1
  while left < right:
   mid = (left + right) // 2
   if current_time - self._pool[mid].last_time >= self.idle_timeout:
    left = mid + 1
   else:
    right = mid
  self._pool = self._pool[left:]
  self.conn_num -= left
 self._lock.release()

　　由于我們獲取和釋放連接都是從self._pool的尾部開(kāi)始操作的, 因此self._pool這個(gè)容器是一個(gè)先進(jìn)后出隊(duì)列, 它里面放著的連接, 一定是越靠近頭部的閑置時(shí)間越長(zhǎng), 從頭到尾閑置時(shí)間依次遞減. 基于這個(gè)原因, 我們使用二分法來(lái)找出列表中第一個(gè)沒(méi)有閑置超時(shí)的連接, 然后把在它之前的連接一次性刪除, 這樣就能達(dá)到O(logN)的時(shí)間復(fù)雜度, 算是一種比較高效的方法. 需要注意的是, 如果連接池內(nèi)所有的連接都是超時(shí)的, 那么這種方法是刪不干凈的, 需要對(duì)這種邊界情況單獨(dú)處理.

三. 總結(jié)

1. 完整代碼及分析

　　這個(gè)連接池的完整代碼如下:

import threading
import time
from typing import Any

from client import HTTPConnection, HTTPResponse


class WrapperHTTPConnection:

 def __init__(self, pool: 'HTTPConnectionPool', conn: HTTPConnection) -> None:
  self.pool = pool
  self.conn = conn
  self.response = None
  self.last_time = time.time()
  self.is_available = True

 def __enter__(self) -> 'WrapperHTTPConnection':
  return self

 def __exit__(self, *exit_info: Any) -> None:
  # 如果response沒(méi)讀完并且連接需要復(fù)用,就棄用這個(gè)連接
  if not self.response.will_close and not self.response.is_closed():
   self.close()
  self.pool.release(self)

 def request(self, *args: Any, **kwargs: Any) -> HTTPResponse:
  self.conn.request(*args, **kwargs)
  self.response = self.conn.get_response()
  return self.response

 def close(self) -> None:
  self.conn.close()
  self.is_available = False


class HTTPConnectionPool:

 def __init__(self, host: str, port: int = None, max_size: int = None, idle_timeout: int = None) -> None:
  """
  :param host: pass
  :param port: pass
  :param max_size: 同時(shí)存在的最大連接數(shù), 默認(rèn)None->連接數(shù)無(wú)限,沒(méi)了就創(chuàng)建
  :param idle_timeout: 單個(gè)連接單次最長(zhǎng)空閑時(shí)間,超時(shí)自動(dòng)關(guān)閉,默認(rèn)None->不限時(shí)
  """
  self.host = host
  self.port = port
  self.max_size = max_size
  self.idle_timeout = idle_timeout
  self._lock = threading.Condition()
  self._pool = []
  # 這里的conn_num指的是總連接數(shù),包括其它線程拿出去正在使用的連接
  self.conn_num = 0
  self.is_closed = False
  self._clearer = None
  self.start_clear_conn()

 def acquire(self, blocking: bool = True, timeout: int = None) -> WrapperHTTPConnection:
  if self.is_closed:
   raise ConnectionPoolClosed
  with self._lock:
   if self.max_size is None or not self.is_full():
    # 在還能創(chuàng)建新連接的情況下,如果沒(méi)有空閑連接,直接創(chuàng)建一個(gè)就行了
    if self.is_pool_empty():
     self._put_connection(self._create_connection())
   else:
    # 不能創(chuàng)建新連接的情況下,如果設(shè)置了blocking=False,沒(méi)連接就報(bào)錯(cuò)
    # 否則,就基于timeout進(jìn)行阻塞,直到超時(shí)或者有可用連接為止
    if not blocking:
     if self.is_pool_empty():
      raise EmptyPoolError
    elif timeout is None:
     while self.is_pool_empty():
      self._lock.wait()
    elif timeout < 0:
     raise ValueError("'timeout' must be a non-negative number")
    else:
     end_time = time.time() + timeout
     while self.is_pool_empty():
      remaining = end_time - time.time()
      if remaining <= 0:
       raise EmptyPoolError
      self._lock.wait(remaining)
   # 走到這一步了,池子里一定有空閑連接
   return self._get_connection()

 def release(self, conn: WrapperHTTPConnection) -> None:
  if self.is_closed:
   # 如果這個(gè)連接是在連接池關(guān)閉后才釋放的,那就不用回連接池了,直接放生
   conn.close()
   return
  # 實(shí)際上,python列表的append操作是線程安全的,可以不加鎖
  # 這里調(diào)用鎖是為了通過(guò)notify方法通知其它正在wait的線程:現(xiàn)在有連接可用了
  with self._lock:
   if not conn.is_available:
    # 如果這個(gè)連接不可用了,就應(yīng)該創(chuàng)建一個(gè)新連接放進(jìn)去,因?yàn)榭赡苓€有其它線程在等著連接用
    conn.close()
    self.conn_num -= 1
    conn = self._create_connection()
   self._put_connection(conn)
   self._lock.notify()

 def _get_connection(self) -> WrapperHTTPConnection:
  # 這個(gè)方法會(huì)把連接從_idle_conn移動(dòng)到_used_conn列表中,并返回這個(gè)連接
  try:
   return self._pool.pop()
  except IndexError:
   raise EmptyPoolError

 def _put_connection(self, conn: WrapperHTTPConnection) -> None:
  conn.last_time = time.time()
  self._pool.append(conn)

 def _create_connection(self) -> WrapperHTTPConnection:
  self.conn_num += 1
  return WrapperHTTPConnection(self, HTTPConnection(self.host, self.port))

 def is_pool_empty(self) -> bool:
  # 這里指的是,空閑可用的連接是否為空
  return len(self._pool) == 0

 def is_full(self) -> bool:
  if self.max_size is None:
   return False
  return self.conn_num >= self.max_size

 def close(self) -> None:
  if self.is_closed:
   return
  self.is_closed = True
  self.stop_clear_conn()
  pool, self._pool = self._pool, None
  for conn in pool:
   conn.close()

 def clear_idle_conn(self) -> None:
  if self.is_closed:
   raise ConnectionPoolClosed
  # 這里開(kāi)一個(gè)新線程來(lái)清理空閑連接,避免了阻塞主線程導(dǎo)致的定時(shí)精度出錯(cuò)
  threading.Thread(target=self._clear_idle_conn).start()

 def _clear_idle_conn(self) -> None:
  if not self._lock.acquire(timeout=self.idle_timeout):
   # 因?yàn)槭敲扛魋elf.idle_timeout秒檢查一次
   # 如果過(guò)了self.idle_timeout秒還沒(méi)申請(qǐng)到鎖,下一次都開(kāi)始了,本次也就不用繼續(xù)了
   return
  current_time = time.time()
  if self.is_pool_empty():
   pass
  elif current_time - self._pool[-1].last_time >= self.idle_timeout:
   # 這里處理下面的二分法沒(méi)法處理的邊界情況,即所有連接都閑置超時(shí)的情況
   self.conn_num -= len(self._pool)
   self._pool.clear()
  else:
   # 通過(guò)二分法找出從左往右第一個(gè)不超時(shí)的連接的指針
   left, right = 0, len(self._pool) - 1
   while left < right:
    mid = (left + right) // 2
    if current_time - self._pool[mid].last_time >= self.idle_timeout:
     left = mid + 1
    else:
     right = mid
   self._pool = self._pool[left:]
   self.conn_num -= left
  self._lock.release()

 def start_clear_conn(self) -> None:
  if self.idle_timeout is None:
   # 如果空閑連接的超時(shí)時(shí)間為無(wú)限,那么就不應(yīng)該清理連接
   return
  self.clear_idle_conn()
  self._clearer = threading.Timer(self.idle_timeout, self.start_clear_conn)
  self._clearer.start()

 def stop_clear_conn(self) -> None:
  if self._clearer is not None:
   self._clearer.cancel()

 def __enter__(self) -> 'HTTPConnectionPool':
  return self

 def __exit__(self, *exit_info: Any) -> None:
  self.close()


class EmptyPoolError(Exception):
 pass


class ConnectionPoolClosed(Exception):
 pass

　　首先, 這個(gè)連接池的核心就是對(duì)連接進(jìn)行管理, 而這包含取出連接和釋放連接兩個(gè)過(guò)程. 因此這東西的本質(zhì)就是一個(gè)生產(chǎn)者消費(fèi)者模型, 取出線程時(shí)是消費(fèi)者, 放入線程時(shí)是生產(chǎn)者, 使用threading自帶的Condition對(duì)象就能完美解決線程安全問(wèn)題, 使二者協(xié)同合作.

　　解決獲取連接和釋放連接這個(gè)問(wèn)題之后, 其實(shí)這個(gè)連接池就已經(jīng)能用了. 但是如果涉及到更多細(xì)節(jié)方面的東西, 比如判斷連接是否可用, 自動(dòng)釋放連接, 清理閑置連接等等, 就需要對(duì)這個(gè)連接進(jìn)行封裝, 為它添加更多的屬性和方法, 這就引入了WrapperHTTPConnection這個(gè)類. 實(shí)現(xiàn)它的__enter___和__exit__方法之后, 就能使用上下文管理器來(lái)自動(dòng)釋放連接. 至于清理閑置連接, 通過(guò)last_time屬性記錄每個(gè)連接的最后釋放時(shí)間, 然后在連接池中添加一個(gè)定時(shí)任務(wù)就行了.

以上就是如何用python實(shí)現(xiàn)一個(gè)HTTP連接池的詳細(xì)內(nèi)容，更多關(guān)于python 實(shí)現(xiàn)一個(gè)HTTP連接池的資料請(qǐng)關(guān)注腳本之家其它相關(guān)文章！

您可能感興趣的文章: