国产成人精品久久免费动漫-国产成人精品天堂-国产成人精品区在线观看-国产成人精品日本-a级毛片无码免费真人-a级毛片毛片免费观看久潮喷

您的位置:首頁(yè)技術(shù)文章
文章詳情頁(yè)

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

瀏覽:3日期:2022-06-29 18:40:36
一. 連接池的原理

首先, 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)用就行了.

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

一個(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è)管理器就是連接池.

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

二. 實(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 EmptyPoolErrordef _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() + timeoutwhile 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 = Noneclass 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 threadingimport timefrom typing import Anyfrom client import HTTPConnection, HTTPResponseclass 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 = Falseclass 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 EmptyPoolErrorelif 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 -= 1conn = 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) // 2if current_time - self._pool[mid].last_time >= self.idle_timeout: left = mid + 1else: 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): passclass 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)注好吧啦網(wǎng)其它相關(guān)文章!

標(biāo)簽: Python 編程
相關(guān)文章:
主站蜘蛛池模板: 亚洲国产精品综合久久一线 | 俄罗斯一级成人毛片 | 97在线国产视频 | 欧美特黄一片aa大片免费看 | 九九精品免视看国产成人 | 成人免费一级毛片在线播放视频 | 丰满寡妇一级毛片 | 国内成人自拍视频 | 国产欧美综合精品一区二区 | 韩国美女一级毛片 | 久久国产毛片 | 欧美视频在线观看免费精品欧美视频 | 国产三级香港三韩国三级 | 成人黄色一级毛片 | avhd101天天看新片 | 台湾黄三级高清在线观看播放 | 99久久免费午夜国产精品 | 奇米888四色在线精品 | 91久久香蕉国产线看 | 日日碰碰 | 国产精品性视频免费播放 | 欧美人与zoxxxx另类9 | 美女午夜色视频在线观看 | 成 人 黄 色 视频 免费观看 | 老少配性xxxxxx | 国产成人午夜片在线观看 | 国外成人在线视频 | 欧美一a级做爰 | 国产日韩欧美一区二区三区在线 | 欧美日韩在线视频不卡一区二区三区 | 特级做人爱c级特级aav毛片 | 日本三级久久 | 中文字幕精品视频 | 免费一级欧美片在线观看 | 国产精品亚洲一区在线播放 | 亚洲欧洲精品国产二码 | 亚洲三级黄色片 | 国产免费播放一区二区 | 欧美手机手机在线视频一区 | 成 人 黄 色 激 情视频网站 | 99久久国产免费福利 |