DBUtils 是Python 的一個用於實現數據庫鏈接池的模塊。mysql
此鏈接池有兩種鏈接模式:sql
DBUtils :提供兩種外部接口:數據庫
PersistentDB :提供線程專用的數據庫鏈接,並自動管理鏈接。session
PooledDB :提供線程間可共享的數據庫鏈接,並自動管理鏈接。ide
爲每一個線程建立一個鏈接,線程即便調用了 close 方法,也不會關閉,只是把連接從新放到連接池,供本身線程再次使用,當線程終止時,連接自動關閉。工具
from DBUtils.PersistentDB import PersistentDB import pymysql POOL = PersistentDB( creator=pymysql, # 使用連接數據庫的模塊 maxusage=None, # 一個連接最多被重複使用的次數,None表示無限制 setsession=[], # 開始會話前執行的命令列表。 ping=0, # ping MySQL服務端,檢查是否服務可用。 closeable=False, # 若是爲False時, conn.close() 實際上被忽略,供下次使用,再線程關閉時,纔會自動關閉連接。若是爲True時, conn.close()則關閉連接,那麼再次調用pool.connection時就會報錯,由於已經真的關閉了鏈接(pool.steady_connection()能夠獲取一個新的連接) threadlocal=None, # 本線程獨享值得對象,用於保存連接對象,若是連接對象被重置 host='127.0.0.1', port=3306, user='root', password='123456', database='test', charset='utf8' ) def func(): conn = POOL.connection(shareable=False) cursor = conn.cursor() cursor.execute('select * from user') result = cursor.fetchall() print(result) cursor.close() conn.close() if __name__ == '__main__': func()
建立一批鏈接到鏈接池,供全部線程共享使用。fetch
import pymysql from DBUtils.PooledDB import PooledDB POOL = PooledDB( creator=pymysql, # 使用連接數據庫的模塊 maxconnections=6, # 鏈接池容許的最大鏈接數,0和None表示不限制鏈接數 mincached=2, # 初始化時,連接池中至少建立的空閒的連接,0表示不建立 maxcached=5, # 連接池中最多閒置的連接,0和None不限制 maxshared=3, # 連接池中最多共享的連接數量,0和None表示所有共享。PS: 無用,由於pymysql和MySQLdb等模塊的 threadsafety都爲1,全部值不管設置爲多少,_maxcached永遠爲0,因此永遠是全部連接都共享。 blocking=True, # 鏈接池中若是沒有可用鏈接後,是否阻塞等待。True,等待;False,不等待而後報錯 maxusage=None, # 一個連接最多被重複使用的次數,None表示無限制 setsession=[], # 開始會話前執行的命令列表。 ping=0, # ping MySQL服務端,檢查是否服務可用。 host='127.0.0.1', port=3306, user='root', password='123456', database='test', charset='utf8' ) def func(): # 檢測當前正在運行鏈接數的是否小於最大連接數,若是不小於則等待或報raise TooManyConnections異常 # 不然則優先去初始化時建立的連接中獲取連接 SteadyDBConnection。 # 而後將SteadyDBConnection對象封裝到PooledDedicatedDBConnection中並返回。 # 若是最開始建立的連接沒有連接,則去建立一個SteadyDBConnection對象,再封裝到PooledDedicatedDBConnection中並返回。 # 一旦關閉連接後,鏈接就返回到鏈接池讓後續線程繼續使用。 conn = POOL.connection() cursor = conn.cursor(pymysql.cursors.DictCursor) cursor.execute('select * from user') result = cursor.fetchall() print(result) conn.close() if __name__ == '__main__': func()
# TEST數據庫信息 DB_TEST_HOST = "127.0.0.1" DB_TEST_PORT = 3306 DB_TEST_DBNAME = "1019" DB_TEST_USER = "root" DB_TEST_PASSWORD = "root" # 數據庫鏈接編碼 DB_CHARSET = "utf8" # mincached : 啓動時開啓的閒置鏈接數量(缺省值 0 開始時不建立鏈接) DB_MIN_CACHED = 10 # maxcached : 鏈接池中容許的閒置的最多鏈接數量(缺省值 0 表明不閒置鏈接池大小) DB_MAX_CACHED = 10 # maxshared : 共享鏈接數容許的最大數量(缺省值 0 表明全部鏈接都是專用的)若是達到了最大數量,被請求爲共享的鏈接將會被共享使用 DB_MAX_SHARED = 20 # maxconnecyions : 建立鏈接池的最大數量(缺省值 0 表明不限制) DB_MAX_CONNECYIONS = 100 # blocking : 設置在鏈接池達到最大數量時的行爲(缺省值 0 或 False 表明返回一個錯誤<toMany......>; 其餘表明阻塞直到鏈接數減小,鏈接被分配) DB_BLOCKING = True # maxusage : 單個鏈接的最大容許複用次數(缺省值 0 或 False 表明不限制的複用).當達到最大數時,鏈接會自動從新鏈接(關閉和從新打開) DB_MAX_USAGE = 0 # setsession : 一個可選的SQL命令列表用於準備每一個會話,如["set datestyle to german", ...] DB_SET_SESSION = None
import pymysql from DBUtils.PooledDB import PooledDB import db_config as Config class ConnectionPool(object): __pool = None def __enter__(self): self.conn = self.__getConn() self.cursor = self.conn.cursor() print("數據庫建立conn和cursor") return self def __getConn(self): if self.__pool is None: self.__pool = PooledDB(creator=pymysql, mincached=Config.DB_MIN_CACHED, maxcached=Config.DB_MAX_CACHED, maxshared=Config.DB_MAX_SHARED, maxconnections=Config.DB_MAX_CONNECYIONS, blocking=Config.DB_BLOCKING, maxusage=Config.DB_MAX_USAGE, setsession=Config.DB_SET_SESSION, host=Config.DB_TEST_HOST, port=Config.DB_TEST_PORT, user=Config.DB_TEST_USER, passwd=Config.DB_TEST_PASSWORD, db=Config.DB_TEST_DBNAME, use_unicode=False, charset=Config.DB_CHARSET) return self.__pool.connection() def __exit__(self, type, value, trace): """ @summary: 釋放鏈接池資源 """ self.cursor.close() self.conn.close() print("鏈接池釋放conn和cursor") def getconn(self): ''' 從鏈接池中取出一個鏈接 ''' conn = self.__getConn() cursor = conn.cursor(pymysql.cursors.DictCursor) return cursor, conn def close(self): ''' 關閉鏈接歸還給鏈接池 ''' self.cursor.close() self.conn.close() print("鏈接池釋放conn和cursor") POOL = ConnectionPool() class MysqlHelper(object): mysql = None def __init__(self): self.db = POOL def __new__(cls, *args, **kwargs): if not hasattr(cls, 'inst'): cls.inst = super(MysqlHelper, cls).__new__(cls, *args, **kwargs) return cls.inst def selectall(self, sql='', param=()): ''' 查詢全部 ''' try: cursor, conn = self.execute(sql, param) res = cursor.fetchall() self.close(cursor, conn) return res except Exception as e: print('selectall except ', e.args) self.close(cursor, conn) return None def selectone(self, sql='', param=()): ''' 查詢一條 ''' try: cursor, conn = self.execute(sql, param) res = cursor.fetchone() self.close(cursor, conn) return res except Exception as e: print('selectone except ', e.args) self.close(cursor, conn) return None def insert(self, sql='', param=()): ''' 增長 ''' try: cursor, conn = self.execute(sql, param) print('============') _id = cursor.lastrowid print('_id ', _id) conn.commit() self.close(cursor, conn) # 防止表中沒有id返回0 if _id == 0: return True return _id except Exception as e: print('insert except ', e.args) conn.rollback() self.close(cursor, conn) # self.conn.rollback() return 0 def insertmany(self, sql='', param=()): ''' 增長多行 ''' cursor, conn = self.db.getconn() try: cursor.executemany(sql, param) conn.commit() self.close(cursor, conn) return True except Exception as e: print('insert many except ', e.args) conn.rollback() self.close(cursor, conn) return False def delete(self, sql='', param=()): ''' 刪除 ''' try: cursor, conn = self.execute(sql, param) self.close(cursor, conn) return True except Exception as e: print('delete except ', e.args) conn.rollback() self.close(cursor, conn) return False def update(self, sql='', param=()): ''' 更新 ''' try: cursor, conn = self.execute(sql, param) self.close(cursor, conn) return True except Exception as e: print('update except ', e.args) conn.rollback() self.close(cursor, conn) return False @classmethod def getInstance(self): if MysqlHelper.mysql == None: MysqlHelper.mysql = MysqlHelper() return MysqlHelper.mysql # 執行命令 def execute(self, sql='', param=(), autoclose=False): cursor, conn = self.db.getconn() try: if param: cursor.execute(sql, param) else: cursor.execute(sql) conn.commit() if autoclose: self.close(cursor, conn) except Exception as e: pass return cursor, conn def executemany(self, list=[]): ''' # 執行多條命令 '[{"sql":"xxx","param":"xx"}....]' ''' cursor, conn = self.db.getconn() try: for order in list: sql = order['sql'] param = order['param'] if param: cursor.execute(sql, param) else: cursor.execute(sql) conn.commit() self.close(cursor, conn) return True except Exception as e: print('execute failed========', e.args) conn.rollback() self.close(cursor, conn) return False def close(self, cursor, conn): cursor.close() conn.close() print("PT鏈接池釋放con和cursor")