20 flask之資料庫連線池DBUtils
阿新 • • 發佈:2022-03-24
DBUtils是Python的一個用於實現資料庫連線池的模組
此連線池有兩種連線模式:
DBUtils提供兩種外部介面:
PersistentDB :提供執行緒專用的資料庫連線,並自動管理連線。
PooledDB :提供執行緒間可共享的資料庫連線,並自動管理連線。
模式一
為每個執行緒建立一個連線,執行緒即使呼叫了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()
模式二
模式二:建立一批連線到連線池,供所有執行緒共享使用。
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() # print(th, '連結被拿走了', conn1._con) # print(th, '池子裡目前有', pool._idle_cache, '\r\n') cursor = conn.cursor() cursor.execute('select * from user') result = cursor.fetchall() print(result) conn.close() if __name__ == '__main__': func()