python實現資料庫連線池

2021-09-18 06:11:12 字數 2955 閱讀 5904

以下python**實現了乙個簡單的資料庫連線池,以sqlite3為例

class pool(object):

"""乙個資料庫連線池"""

def __init__(self, max_active=5, max_wait=none, init_size=0, db_type="sqlite3", **config):

self.__freeconns = queue(max_active)

self.maxwait = max_wait

self.db_type = db_type

self.config = config

if init_size > max_active:

init_size = max_active

for i in range(init_size):

self.free(self._create_conn())

def __del__(self):

print("__del__ pool..")

self.release()

def release(self):

"""釋放資源,關閉池中的所有連線"""

while self.__freeconns and not self.__freeconns.empty():

con = self.get()

con.release()

self.__freeconns = none

def _create_conn(self):

"""建立連線 """

if self.db_type in dbcs:

return dbcs[self.db_type](**self.config)

def get(self, timeout=none):

"""獲取乙個連線

@param timeout:超時時間

"""if timeout is none:

timeout = self.maxwait

conn = none

if self.__freeconns.empty(): # 如果容器是空的,直接建立乙個連線

conn = self._create_conn()

else:

conn = self.__freeconns.get(timeout=timeout)

conn.pool = self

conn.row_factory = sqlite3.row

return conn

def free(self, conn):

"""將乙個連線放回池中

@param conn: 連線物件

"""conn.pool = none

if self.__freeconns.full(): # 如果當前連線池已滿,直接關閉連線

conn.release()

return

self.__freeconns.put_nowait(conn)

class poolingconnection(object):

__metaclass__ = abcmeta

def __init__(self, **config):

self.conn = none

self.config = config

self.pool = none

def __del__(self):

self.release()

def __enter__(self):

pass

def __exit__(self, exc_type, exc_value, traceback):

self.close()

def release(self):

print("release poolingconnection..")

if self.conn is not none:

self.conn.close()

self.conn = none

self.pool = none

def close(self):

if self.pool is none:

raise poolexception("連線已關閉")

self.pool.free(self)

def __getattr__(self, val):

if self.conn is none and self.pool is not none:

self.conn = self._create_conn(**self.config)

if self.conn is none:

raise poolexception("無法建立資料庫連線或連線已關閉")

return getattr(self.conn, val)

@abstractmethod //定義乙個抽象方法,該方法只有在子類中例項化才生效

def _create_conn(self, **config):

pass

class sqlit3poolconnection(poolingconnection):

def _create_conn(self, **config):

conn = sqlite3.connect(**config)

conn.row_factory = sqlite3.row

return conn

dbcs =

pool = pool(database=os.path.join(get_db_path(), "cluster.db3"))

在需要的地方呼叫該pool物件,

使用pool.get(5)獲取乙個conn物件

使用完成後務必關閉conn連線,因為sqlite不支援跨執行緒操作,若不關閉則可能會發生異常

以上**只是乙個簡單的實現,有需要的童鞋可根據具體需求自行改造,希望對大家有幫助。

python 資料庫連線池

from dbutils.pooleddb import pooleddb import pymysql pool pooleddb creator pymysql,使用鏈結資料庫的模組 maxconnections 6,連線池允許的最大連線數,0和none表示不限制連線數 mincached 2,...

資料庫連線池 Redis連線池

基本原理 在內部物件池中,維護一定數量的資料庫連線,並對外暴露資料庫連線的獲取和返回方法。如外部使用者可通過getconnection方法獲取資料庫連線,使用完畢後再通過releaseconnection方法將連線返回,注意此時的連線並沒有關閉,而是由連線池管理器 並為下一次使用做好準備。2.作用 ...

資料庫連線池

實現資料連線池,讓系統有更高有執行效率 using system using system.data using system.data.sqlclient using system.collections using system.threading public class dataaccess...