一文讀懂連接池技術(shù)原理、設(shè)計與實現(xiàn)

作者：曹金龍

概述

連接池的作用就是為了提高性能，將已經(jīng)創(chuàng)建好的連接保存在池中，當(dāng)有請求來時，直接使用已經(jīng)創(chuàng)建好的連接對Server端進行訪問。這樣省略了創(chuàng)建連接和銷毀連接的過程（TCP連接建立時的三次握手和銷毀時的四次握手），從而在性能上得到了提高。

連接池設(shè)計的基本原理是這樣的：

說的通俗點，可以把連接池理解為一個一個的管道，在管道空閑時，便可以取出使用；同時，也可以鋪設(shè)新的管道（當(dāng)然不能超過最大連接數(shù)的限制）。使用完之后，管道就變?yōu)榭臻e了。

通常比較常用的連接池是數(shù)據(jù)庫連接池，HTTP Client連接池，我也自己編寫過連接池，如Thrift連接池及插入Rabbitmq隊列的連接池。

下面分析三個典型的連接池的設(shè)計。

數(shù)據(jù)庫連接池

首先剖析一下數(shù)據(jù)庫連接池的設(shè)計與實現(xiàn)的原理。DBUtils 屬于數(shù)據(jù)庫連接池實現(xiàn)模塊，用于連接DB-API 2模塊，對數(shù)據(jù)庫連接線程化，使可以安全和高效的訪問數(shù)據(jù)庫的模塊。本文主要分析一下PooledDB的流程。

DBUtils.PooledDB使用DB-API 2模塊實現(xiàn)了一個強硬的、線程安全的、有緩存的、可復(fù)用的數(shù)據(jù)庫連接。

如下圖展示了使用PooledDB時的工作流程：

本文主要考慮dedicated connections，即專用數(shù)據(jù)庫連接，在初始化時連接池時，就需要指定mincached、maxcached以及maxconnections等參數(shù)，分別表示連接池的最小連接數(shù)、連接池的最大連接數(shù)以及系統(tǒng)可用的最大連接數(shù)，同時，blocking參數(shù)表征了當(dāng)獲取不到連接的時候是阻塞等待獲取連接還是返回異常：

if not blocking: ? ?def wait(): ? ? ? ?raise TooManyConnections ? ?self._condition.wait = wait

在連接池初始化時，就會建立mincached個連接，代碼如下：

# Establish an initial number of idle database connections:idle = [self.dedicated_connection() for i in range(mincached)]while idle: ? ?idle.pop().close()

里面有close方法，看一下連接close方法的實現(xiàn)：

def close(self): ? ?"""Close the pooled dedicated connection.""" ? ?# Instead of actually closing the connection, ? ?# return it to the pool for future reuse. ? ?if self._con: ? ? ? ?self._pool.cache(self._con) ? ? ? ?self._con = None

主要是實現(xiàn)了cache方法，看一下具體代碼：

def cache(self, con): ? ?"""Put a dedicated connection back into the idle cache.""" ? ?self._condition.acquire() ? ?try: ? ? ? ?if not self._maxcached or len(self._idle_cache) < self._maxcached: ? ? ? ? ? ?con._reset(force=self._reset) # rollback possible transaction ? ? ? ? ? ?# the idle cache is not full, so put it there ? ? ? ? ? ?self._idle_cache.append(con) # append it to the idle cache ? ? ? ?else: # if the idle cache is already full, ? ? ? ? ? ?con.close() # then close the connection ? ? ? ?self._connections -= 1 ? ? ? ?self._condition.notify() ? ?finally: ? ? ? ?self._condition.release()

由上述代碼可見，close并不是把連接關(guān)閉，而是在連接池的數(shù)目小于maxcached的時候，將連接放回連接池，而大于此值時，關(guān)閉該連接。同時可以注意到，在放回連接池之前，需要將事務(wù)進行回滾，避免在使用連接池的時候有存活的事務(wù)沒有提交。這可以保證進入連接池的連接都是可用的。

而獲取連接的過程正如之前討論的，先從連接池中獲取連接，如果獲取連接失敗，則新建立連接：

# try to get a dedicated connection ? ?self._condition.acquire() ? ?try: ? ? ? ?while (self._maxconnections ? ? ? ? ? ? ? ?and self._connections >= self._maxconnections): ? ? ? ? ? ?self._condition.wait() ? ? ? ?# connection limit not reached, get a dedicated connection ? ? ? ?try: # first try to get it from the idle cache ? ? ? ? ? ?con = self._idle_cache.pop(0) ? ? ? ?except IndexError: # else get a fresh connection ? ? ? ? ? ?con = self.steady_connection() ? ? ? ?else: ? ? ? ? ? ?con._ping_check() # check connection ? ? ? ?con = PooledDedicatedDBConnection(self, con) ? ? ? ?self._connections += 1 ? ?finally: ? ? ? ?self._condition.release()

關(guān)閉連接正如剛剛創(chuàng)建mincached個連接后關(guān)閉連接的流程，在連接池的數(shù)目小于maxcached的時候，將連接放回連接池，而大于此值時，關(guān)閉該連接。

RabbitMQ隊列插入消息連接池

異步消息傳遞是高并發(fā)系統(tǒng)常用的一種技術(shù)手段。而這其中就少不了消息隊列。頻繁的向消息隊列里面插入消息，建立連接釋放連接會是比較大的開銷。所以，可以使用連接池來提高系統(tǒng)性能。

連接池的設(shè)計實現(xiàn)如下：

在獲取連接的時候，先從隊列里面獲取連接，如果獲取不到，則新建立一個連接，如果不能新建立連接，則根據(jù)超時時間，阻塞等待從隊列里面獲取鏈接。如果沒成功，則做最后的嘗試，重新建立連接。代碼實現(xiàn)如下：

 ? ?def get_connection_pipe(self): ? ? ? ?""" ? ? ? ?獲取連接 ? ? ? ?:return: ? ? ? ?""" ? ? ? ?try: ? ? ? ? ? ?connection_pipe = self._queue.get(False) ? ? ? ?except Queue.Empty: ? ? ? ? ? ?try: ? ? ? ? ? ? ? ?connection_pipe = self.get_new_connection_pipe() ? ? ? ? ? ?except GetConnectionException: ? ? ? ? ? ? ? ?timeout = self.timeout ? ? ? ? ? ? ? ?try: ? ? ? ? ? ? ? ? ? ?connection_pipe = self._queue.get(timeout=timeout) ? ? ? ? ? ? ? ?except Queue.Empty: ? ? ? ? ? ? ? ? ? ?try: ? ? ? ? ? ? ? ? ? ? ? ?connection_pipe = self.get_new_connection_pipe() ? ? ? ? ? ? ? ? ? ?except GetConnectionException: ? ? ? ? ? ? ? ? ? ? ? ?logging.error("Too much connections, Get Connection Timeout!") ? ? ? ?if (time.time() - connection_pipe.use_time) > self.disable_time: ? ? ? ? ? ?self.close(connection_pipe) ? ? ? ? ? ?return self.get_connection_pipe() ? ? ? ?return connection_pipe

一個RabbitMQ插入消息隊列的完整連接池設(shè)計如下：

# coding:utf-8import loggingimport threadingimport Queuefrom kombu import Connectionimport timeclass InsertQueue(): ? ?def __init__(self, host=None, port=None, virtual_host=None, heartbeat_interval=3, name=None, password=None, ? ? ? ? ? ? ? ? logger=None, maxIdle=10, maxActive=50, timeout=30, disable_time=20): ? ? ? ?""" ? ? ? ?:param str host: Hostname or IP Address to connect to ? ? ? ?:param int port: TCP port to connect to ? ? ? ?:param str virtual_host: RabbitMQ virtual host to use ? ? ? ?:param int heartbeat_interval: ?How often to send heartbeats ? ? ? ?:param str name: auth credentials name ? ? ? ?:param str password: auth credentials password ? ? ? ?""" ? ? ? ?self.logger = logging if logger is None else logger ? ? ? ?self.host = host ? ? ? ?self.port = port ? ? ? ?self.virtual_host = virtual_host ? ? ? ?self.heartbeat_interval = heartbeat_interval ? ? ? ?self.name = name ? ? ? ?self.password = password ? ? ? ?self.mutex = threading.RLock() ? ? ? ?self.maxIdle = maxIdle ? ? ? ?self.maxActive = maxActive ? ? ? ?self.available = self.maxActive ? ? ? ?self.timeout = timeout ? ? ? ?self._queue = Queue.Queue(maxsize=self.maxIdle) ? ? ? ?self.disable_time = disable_time ? ?def get_new_connection_pipe(self): ? ? ? ?""" ? ? ? ?產(chǎn)生新的隊列連接 ? ? ? ?:return: ? ? ? ?""" ? ? ? ?with self.mutex: ? ? ? ? ? ?if self.available <= 0: ? ? ? ? ? ? ? ?raise GetConnectionException ? ? ? ? ? ?self.available -= 1 ? ? ? ?try: ? ? ? ? ? ?conn = Connection(hostname=self.host, ? ? ? ? ? ? ? ? ? ? ? ? ? ? ?port=self.port, ? ? ? ? ? ? ? ? ? ? ? ? ? ? ?virtual_host=self.virtual_host, ? ? ? ? ? ? ? ? ? ? ? ? ? ? ?heartbeat=self.heartbeat_interval, ? ? ? ? ? ? ? ? ? ? ? ? ? ? ?userid=self.name, ? ? ? ? ? ? ? ? ? ? ? ? ? ? ?password=self.password) ? ? ? ? ? ?producer = conn.Producer() ? ? ? ? ? ?return ConnectionPipe(conn, producer) ? ? ? ?except: ? ? ? ? ? ?with self.mutex: ? ? ? ? ? ? ? ?self.available += 1 ? ? ? ? ? ?raise GetConnectionException ? ?def get_connection_pipe(self): ? ? ? ?""" ? ? ? ?獲取連接 ? ? ? ?:return: ? ? ? ?""" ? ? ? ?try: ? ? ? ? ? ?connection_pipe = self._queue.get(False) ? ? ? ?except Queue.Empty: ? ? ? ? ? ?try: ? ? ? ? ? ? ? ?connection_pipe = self.get_new_connection_pipe() ? ? ? ? ? ?except GetConnectionException: ? ? ? ? ? ? ? ?timeout = self.timeout ? ? ? ? ? ? ? ?try: ? ? ? ? ? ? ? ? ? ?connection_pipe = self._queue.get(timeout=timeout) ? ? ? ? ? ? ? ?except Queue.Empty: ? ? ? ? ? ? ? ? ? ?try: ? ? ? ? ? ? ? ? ? ? ? ?connection_pipe = self.get_new_connection_pipe() ? ? ? ? ? ? ? ? ? ?except GetConnectionException: ? ? ? ? ? ? ? ? ? ? ? ?logging.error("Too much connections, Get Connection Timeout!") ? ? ? ?if (time.time() - connection_pipe.use_time) > self.disable_time: ? ? ? ? ? ?self.close(connection_pipe) ? ? ? ? ? ?return self.get_connection_pipe() ? ? ? ?return connection_pipe ? ?def close(self, connection_pipe): ? ? ? ?""" ? ? ? ?close the connection and the correlative channel ? ? ? ?:param connection_pipe: ? ? ? ?:return: ? ? ? ?""" ? ? ? ?with self.mutex: ? ? ? ? ? ?self.available += 1 ? ? ? ? ? ?connection_pipe.close() ? ? ? ?return ? ?def insert_message(self, exchange=None, body=None, routing_key='', mandatory=True): ? ? ? ?""" ? ? ? ?insert message to queue ? ? ? ?:param str exchange: exchange name ? ? ? ?:param str body: message ? ? ? ?:param str routing_key: routing key ? ? ? ?:param bool mandatory: is confirm: True means confirm, False means not confirm ? ? ? ?:return: ? ? ? ?""" ? ? ? ?put_into_queue_flag = True ? ? ? ?insert_result = False ? ? ? ?connection_pipe = None ? ? ? ?try: ? ? ? ? ? ?connection_pipe = self.get_connection_pipe() ? ? ? ? ? ?producer = connection_pipe.channel ? ? ? ? ? ?use_time = time.time() ? ? ? ? ? ?producer.publish(exchange=exchange, ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? body=body, ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? delivery_mode=2, ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? routing_key=routing_key, ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? mandatory=mandatory ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ) ? ? ? ? ? ?insert_result = True ? ? ? ?except Exception: ? ? ? ? ? ?insert_result = False ? ? ? ? ? ?put_into_queue_flag = False ? ? ? ?finally: ? ? ? ? ? ?if put_into_queue_flag is True: ? ? ? ? ? ? ? ?try: ? ? ? ? ? ? ? ? ? ?connection_pipe.use_time = use_time ? ? ? ? ? ? ? ? ? ?self._queue.put_nowait(connection_pipe) ? ? ? ? ? ? ? ?except Queue.Full: ? ? ? ? ? ? ? ? ? ?self.close(connection_pipe) ? ? ? ? ? ?else: ? ? ? ? ? ? ? ?if connection_pipe is not None: ? ? ? ? ? ? ? ? ? ?self.close(connection_pipe) ? ? ? ?return insert_resultclass ConnectionPipe(object): ? ?""" ? ?connection和channel對象的封裝 ? ?""" ? ?def __init__(self, connection, channel): ? ? ? ?self.connection = connection ? ? ? ?self.channel = channel ? ? ? ?self.use_time = time.time() ? ?def close(self): ? ? ? ?try: ? ? ? ? ? ?self.connection.close() ? ? ? ?except Exception as ex: ? ? ? ? ? ?passclass GetConnectionException(): ? ?""" ? ?獲取連接異常 ? ?""" ? ?pass

Thrift連接池

Thrift是什么呢？簡而言之，Thrift定義一個簡單的文件，包含數(shù)據(jù)類型和服務(wù)接口，以作為輸入文件，編譯器生成代碼用來方便地生成RPC客戶端和服務(wù)器通信的方式。實際上就是一種遠程調(diào)用的方式，因為協(xié)議棧為TCP層，所以相對于HTTP層效率會更高。

Thrift連接池的設(shè)計同數(shù)據(jù)庫連接池類似，流程圖如下：

思路依舊是，在獲取連接時，先從連接池中獲取連接，若池中無連接，則判斷是否可以新建連接，若不能新建連接，則阻塞等待連接。

在從池中獲取不到隊列的時候的處理方式，本設(shè)計處理方式為：當(dāng)獲取不到連接時，將這部分請求放入一個等待隊列，等待獲取連接；而當(dāng)關(guān)閉連接放回連接池時，優(yōu)先判斷這個隊列是否有等待獲取連接的請求，若有，則優(yōu)先分配給這些請求。

獲取不到連接時處理代碼如下，將請求放入一個隊列進行阻塞等待獲取連接：

async_result = AsyncResult()self.no_client_queue.appendleft(async_result)client = async_result.get() ?# blocking

而當(dāng)有連接釋放需要放回連接池時，需要優(yōu)先考慮這部分請求，代碼如下：

def put_back_connections(self, client): ? ?""" ? ?線程安全 ? ?將連接放回連接池，邏輯如下： ? ?1、如果有請求尚未獲取到連接，請求優(yōu)先 ? ?2、如果連接池中的連接的數(shù)目小于maxIdle，則將該連接放回連接池 ? ?3、關(guān)閉連接 ? ?:param client: ? ?:return: ? ?""" ? ?with self.lock: ? ? ? ?if self.no_client_queue.__len__() > 0: ? ? ? ? ? ?task = self.no_client_queue.pop() ? ? ? ? ? ?task.set(client) ? ? ? ?elif self.connections.__len__() < self.maxIdle: ? ? ? ? ? ?self.connections.add(client) ? ? ? ?else: ? ? ? ? ? ?client.close() ? ? ? ? ? ?self.pool_size -= 1

最后，基于thrift連接池，介紹一個簡單的服務(wù)化框架的實現(xiàn)。

服務(wù)化框架分為兩部分：RPC、注冊中心。

通常的架構(gòu)圖為：

通過Thrift連接池作為客戶端，而Zookeeper作為注冊中心，設(shè)計服務(wù)框架。具體就是服務(wù)端在啟動服務(wù)的時候到Zookeeper進行注冊，而客戶端在啟動的時候通過Zookeeper發(fā)現(xiàn)服務(wù)端的IP和端口，通過Thrift連接池輪詢建立連接訪問服務(wù)端的服務(wù)。

具體設(shè)計的代碼如下，代碼有點長，細(xì)細(xì)研讀一定有所收獲的：

# coding: utf-8import threadingfrom collections import dequeimport loggingimport socketimport timefrom kazoo.client import KazooClientfrom thriftpy.protocol import TBinaryProtocolFactoryfrom thriftpy.transport import ( ? ?TBufferedTransportFactory, ? ?TSocket,)from gevent.event import AsyncResultfrom gevent import Timeoutfrom error import CTECThriftClientErrorfrom thriftpy.thrift import TClientfrom thriftpy.transport import TTransportExceptionclass ClientPool: ? ?def __init__(self, service, server_hosts=None, zk_path=None, zk_hosts=None, logger=None, max_renew_times=3, maxActive=20, ? ? ? ? ? ? ? ? maxIdle=10, get_connection_timeout=30, socket_timeout=30, disable_time=3): ? ? ? ?""" ? ? ? ?:param service: Thrift的Service名稱 ? ? ? ?:param server_hosts: 服務(wù)提供者地址，數(shù)組類型，['ip:port','ip:port'] ? ? ? ?:param zk_path: 服務(wù)提供者在zookeeper中的路徑 ? ? ? ?:param zk_hosts: zookeeper的host地址，多個請用逗號隔開 ? ? ? ?:param max_renew_times: 最大重連次數(shù) ? ? ? ?:param maxActive: 最大連接數(shù) ? ? ? ?:param maxIdle: 最大空閑連接數(shù) ? ? ? ?:param get_connection_timeout:獲取連接的超時時間 ? ? ? ?:param socket_timeout: 讀取數(shù)據(jù)的超時時間 ? ? ? ?:param disable_time: 連接失效時間 ? ? ? ?""" ? ? ? ?# 負(fù)載均衡隊列 ? ? ? ?self.load_balance_queue = deque() ? ? ? ?self.service = service ? ? ? ?self.lock = threading.RLock() ? ? ? ?self.max_renew_times = max_renew_times ? ? ? ?self.maxActive = maxActive ? ? ? ?self.maxIdle = maxIdle ? ? ? ?self.connections = set() ? ? ? ?self.pool_size = 0 ? ? ? ?self.get_connection_timeout = get_connection_timeout ? ? ? ?self.no_client_queue = deque() ? ? ? ?self.socket_timeout = socket_timeout ? ? ? ?self.disable_time = disable_time ? ? ? ?self.logger = logging if logger is None else logger ? ? ? ?if zk_hosts: ? ? ? ? ? ?self.kazoo_client = KazooClient(hosts=zk_hosts) ? ? ? ? ? ?self.kazoo_client.start() ? ? ? ? ? ?self.zk_path = zk_path ? ? ? ? ? ?self.zk_hosts = zk_hosts ? ? ? ? ? ?# 定義Watcher ? ? ? ? ? ?self.kazoo_client.ChildrenWatch(path=self.zk_path, ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ?func=self.watcher) ? ? ? ? ? ?# 刷新連接池中的連接對象 ? ? ? ? ? ?self.__refresh_thrift_connections(self.kazoo_client.get_children(self.zk_path)) ? ? ? ?elif server_hosts: ? ? ? ? ? ?self.server_hosts = server_hosts ? ? ? ? ? ?# 復(fù)制新的IP地址到負(fù)載均衡隊列中 ? ? ? ? ? ?self.load_balance_queue.extendleft(self.server_hosts) ? ? ? ?else: ? ? ? ? ? ?raise CTECThriftClientError('沒有指定服務(wù)器獲取方式！') ? ?def get_new_client(self): ? ? ? ?""" ? ? ? ?輪詢在每個ip:port的連接池中獲取連接（線程安全） ? ? ? ?從當(dāng)前隊列右側(cè)取出ip:port信息，獲取client ? ? ? ?將連接池對象放回到當(dāng)前隊列的左側(cè) ? ? ? ?請求或連接超時時間，默認(rèn)30秒 ? ? ? ?:return: ? ? ? ?""" ? ? ? ?with self.lock: ? ? ? ? ? ?if self.pool_size < self.maxActive: ? ? ? ? ? ? ? ?try: ? ? ? ? ? ? ? ? ? ?ip = self.load_balance_queue.pop() ? ? ? ? ? ? ? ?except IndexError: ? ? ? ? ? ? ? ? ? ?raise CTECThriftClientError('沒有可用的服務(wù)提供者列表！') ? ? ? ? ? ? ? ?if ip: ? ? ? ? ? ? ? ? ? ?self.load_balance_queue.appendleft(ip) ? ? ? ? ? ? ? ? ? ?# 創(chuàng)建新的thrift client ? ? ? ? ? ? ? ? ? ?t_socket = TSocket(ip.split(':')[0], int(ip.split(':')[1]), ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? socket_timeout=1000 * self.socket_timeout) ? ? ? ? ? ? ? ? ? ?proto_factory = TBinaryProtocolFactory() ? ? ? ? ? ? ? ? ? ?trans_factory = TBufferedTransportFactory() ? ? ? ? ? ? ? ? ? ?transport = trans_factory.get_transport(t_socket) ? ? ? ? ? ? ? ? ? ?protocol = proto_factory.get_protocol(transport) ? ? ? ? ? ? ? ? ? ?transport.open() ? ? ? ? ? ? ? ? ? ?client = TClient(self.service, protocol) ? ? ? ? ? ? ? ? ? ?self.pool_size += 1 ? ? ? ? ? ? ? ?return client ? ? ? ? ? ?else: ? ? ? ? ? ? ? ?return None ? ?def close(self): ? ? ? ?""" ? ? ? ?關(guān)閉所有連接池和zk客戶端 ? ? ? ?:return: ? ? ? ?""" ? ? ? ?if getattr(self, 'kazoo_client', None): ? ? ? ? ? ?self.kazoo_client.stop() ? ?def watcher(self, children): ? ? ? ?""" ? ? ? ?zk的watcher方法，負(fù)責(zé)檢測zk的變化，刷新當(dāng)前雙端隊列中的連接池 ? ? ? ?:param children: 子節(jié)點，即服務(wù)提供方的列表 ? ? ? ?:return: ? ? ? ?""" ? ? ? ?self.__refresh_thrift_connections(children) ? ?def __refresh_thrift_connections(self, children): ? ? ? ?""" ? ? ? ?刷新服務(wù)提供者在當(dāng)前隊列中的連接池信息（線程安全），主要用于zk刷新 ? ? ? ?:param children: ? ? ? ?:return: ? ? ? ?""" ? ? ? ?with self.lock: ? ? ? ? ? ?# 清空負(fù)載均衡隊列 ? ? ? ? ? ?self.load_balance_queue.clear() ? ? ? ? ? ?# 清空連接池 ? ? ? ? ? ?self.connections.clear() ? ? ? ? ? ?# 復(fù)制新的IP地址到負(fù)載均衡隊列中 ? ? ? ? ? ?self.load_balance_queue.extendleft(children) ? ?def __getattr__(self, name): ? ? ? ?""" ? ? ? ?函數(shù)調(diào)用，最大重試次數(shù)為max_renew_times ? ? ? ?:param name: ? ? ? ?:return: ? ? ? ?""" ? ? ? ?def method(*args, **kwds): ? ? ? ? ? ?# 從連接池獲取連接 ? ? ? ? ? ?client = self.get_client_from_pool() ? ? ? ? ? ?# 連接池中無連接 ? ? ? ? ? ?if client is None: ? ? ? ? ? ? ? ?# 設(shè)置獲取連接的超時時間 ? ? ? ? ? ? ? ?time_out = Timeout(self.get_connection_timeout) ? ? ? ? ? ? ? ?time_out.start() ? ? ? ? ? ? ? ?try: ? ? ? ? ? ? ? ? ? ?async_result = AsyncResult() ? ? ? ? ? ? ? ? ? ?self.no_client_queue.appendleft(async_result) ? ? ? ? ? ? ? ? ? ?client = async_result.get() ?# blocking ? ? ? ? ? ? ? ?except: ? ? ? ? ? ? ? ? ? ?with self.lock: ? ? ? ? ? ? ? ? ? ? ? ?if client is None: ? ? ? ? ? ? ? ? ? ? ? ? ? ?self.no_client_queue.remove(async_result) ? ? ? ? ? ? ? ? ? ? ? ? ? ?self.logger.error("Get Connection Timeout!") ? ? ? ? ? ? ? ?finally: ? ? ? ? ? ? ? ? ? ?time_out.cancel() ? ? ? ? ? ?if client is not None: ? ? ? ? ? ? ? ?for i in xrange(self.max_renew_times): ? ? ? ? ? ? ? ? ? ?try: ? ? ? ? ? ? ? ? ? ? ? ?put_back_flag = True ? ? ? ? ? ? ? ? ? ? ? ?client.last_use_time = time.time() ? ? ? ? ? ? ? ? ? ? ? ?fun = getattr(client, name, None) ? ? ? ? ? ? ? ? ? ? ? ?return fun(*args, **kwds) ? ? ? ? ? ? ? ? ? ?except socket.timeout: ? ? ? ? ? ? ? ? ? ? ? ?self.logger.error("Socket Timeout!") ? ? ? ? ? ? ? ? ? ? ? ?# 關(guān)閉連接，不關(guān)閉會導(dǎo)致亂序 ? ? ? ? ? ? ? ? ? ? ? ?put_back_flag = False ? ? ? ? ? ? ? ? ? ? ? ?self.close_one_client(client) ? ? ? ? ? ? ? ? ? ? ? ?break ? ? ? ? ? ? ? ? ? ?except TTransportException, e: ? ? ? ? ? ? ? ? ? ? ? ?put_back_flag = False ? ? ? ? ? ? ? ? ? ? ? ?if e.type == TTransportException.END_OF_FILE: ? ? ? ? ? ? ? ? ? ? ? ? ? ?self.logger.warning("Socket Connection Reset Error,%s", e) ? ? ? ? ? ? ? ? ? ? ? ? ? ?with self.lock: ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ?client.close() ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ?self.pool_size -= 1 ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ?client = self.get_new_client() ? ? ? ? ? ? ? ? ? ? ? ?else: ? ? ? ? ? ? ? ? ? ? ? ? ? ?self.logger.error("Socket Error,%s", e) ? ? ? ? ? ? ? ? ? ? ? ? ? ?self.close_one_client(client) ? ? ? ? ? ? ? ? ? ? ? ? ? ?break ? ? ? ? ? ? ? ? ? ?except socket.error, e: ? ? ? ? ? ? ? ? ? ? ? ?put_back_flag = False ? ? ? ? ? ? ? ? ? ? ? ?if e.errno == socket.errno.ECONNABORTED: ? ? ? ? ? ? ? ? ? ? ? ? ? ?self.logger.warning("Socket Connection aborted Error,%s", e) ? ? ? ? ? ? ? ? ? ? ? ? ? ?with self.lock: ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ?client.close() ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ?self.pool_size -= 1 ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ?client = self.get_new_client() ? ? ? ? ? ? ? ? ? ? ? ?else: ? ? ? ? ? ? ? ? ? ? ? ? ? ?self.logger.error("Socket Error, %s", e) ? ? ? ? ? ? ? ? ? ? ? ? ? ?self.close_one_client(client) ? ? ? ? ? ? ? ? ? ? ? ? ? ?break ? ? ? ? ? ? ? ? ? ?except Exception as e: ? ? ? ? ? ? ? ? ? ? ? ?put_back_flag = False ? ? ? ? ? ? ? ? ? ? ? ?self.logger.error("Thrift Error, %s", e) ? ? ? ? ? ? ? ? ? ? ? ?self.close_one_client(client) ? ? ? ? ? ? ? ? ? ? ? ?break ? ? ? ? ? ? ? ? ? ?finally: ? ? ? ? ? ? ? ? ? ? ? ?# 將連接放回連接池 ? ? ? ? ? ? ? ? ? ? ? ?if put_back_flag is True: ? ? ? ? ? ? ? ? ? ? ? ? ? ?self.put_back_connections(client) ? ? ? ? ? ?return None ? ? ? ?return method ? ?def close_one_client(self, client): ? ? ? ?""" ? ? ? ?線程安全 ? ? ? ?關(guān)閉連接 ? ? ? ?:param client: ? ? ? ?:return: ? ? ? ?""" ? ? ? ?with self.lock: ? ? ? ? ? ?client.close() ? ? ? ? ? ?self.pool_size -= 1 ? ?def put_back_connections(self, client): ? ? ? ?""" ? ? ? ?線程安全 ? ? ? ?將連接放回連接池，邏輯如下： ? ? ? ?1、如果有請求尚未獲取到連接，請求優(yōu)先 ? ? ? ?2、如果連接池中的連接的數(shù)目小于maxIdle，則將該連接放回連接池 ? ? ? ?3、關(guān)閉連接 ? ? ? ?:param client: ? ? ? ?:return: ? ? ? ?""" ? ? ? ?with self.lock: ? ? ? ? ? ?if self.no_client_queue.__len__() > 0: ? ? ? ? ? ? ? ?task = self.no_client_queue.pop() ? ? ? ? ? ? ? ?task.set(client) ? ? ? ? ? ?elif self.connections.__len__() < self.maxIdle: ? ? ? ? ? ? ? ?self.connections.add(client) ? ? ? ? ? ?else: ? ? ? ? ? ? ? ?client.close() ? ? ? ? ? ? ? ?self.pool_size -= 1 ? ?def get_client_from_pool(self): ? ? ? ?""" ? ? ? ?線程安全 ? ? ? ?從連接池中獲取連接，若連接池中有連接，直接取出，否則， ? ? ? ?新建一個連接，若一直無法獲取連接，則返回None ? ? ? ?:return: ? ? ? ?""" ? ? ? ?client = self.get_one_client_from_pool() ? ? ? ?if client is not None and (time.time() - client.last_use_time) < self.disable_time: ? ? ? ? ? ?return client ? ? ? ?else: ? ? ? ? ? ?if client is not None: ? ? ? ? ? ? ? ?self.close_one_client(client) ? ? ? ?client = self.get_new_client() ? ? ? ?if client is not None: ? ? ? ? ? ?return client ? ? ? ?return None ? ?def get_one_client_from_pool(self): ? ? ? ?""" ? ? ? ?線程安全 ? ? ? ?從連接池中獲取一個連接，若取不到連接，則返回None ? ? ? ?:return: ? ? ? ?""" ? ? ? ?with self.lock: ? ? ? ? ? ?if self.connections: ? ? ? ? ? ? ? ?try: ? ? ? ? ? ? ? ? ? ?return self.connections.pop() ? ? ? ? ? ? ? ?except KeyError: ? ? ? ? ? ? ? ? ? ?return None ? ? ? ? ? ?return None

生成海報

免責(zé)聲明：本網(wǎng)站內(nèi)容主要來自原創(chuàng)、合作伙伴供稿和第三方自媒體作者投稿，凡在本網(wǎng)站出現(xiàn)的信息，均僅供參考。本網(wǎng)站將盡力確保所提供信息的準(zhǔn)確性及可靠性，但不保證有關(guān)資料的準(zhǔn)確性及可靠性，讀者在使用前請進一步核實，并對任何自主決定的行為負(fù)責(zé)。本網(wǎng)站對有關(guān)資料所引致的錯誤、不確或遺漏，概不負(fù)任何法律責(zé)任。任何單位或個人認(rèn)為本網(wǎng)站中的網(wǎng)頁或鏈接內(nèi)容可能涉嫌侵犯其知識產(chǎn)權(quán)或存在不實內(nèi)容時，應(yīng)及時向本網(wǎng)站提出書面權(quán)利通知或不實情況說明，并提供身份證明、權(quán)屬證明及詳細(xì)侵權(quán)或不實情況證明。本網(wǎng)站在收到上述法律文件后，將會依法盡快聯(lián)系相關(guān)文章源頭核實，溝通刪除相關(guān)內(nèi)容或斷開相關(guān)鏈接。