以下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...