創建數據庫連接池

?

import time import pymysql import threading from DBUtils.PooledDB import PooledDB, SharedDBConnection 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=[], # 開始會話前執行的命令列表。如：["set datestyle to ...", "set time zone ..."]ping=0,# ping MySQL服務端，檢查是否服務可用。# 如：0 = None = never, 1 = default = whenever it is requested, 2 = when a cursor is created, 4 = when a query is executed, 7 = alwayshost='127.0.0.1',port=3306,user='root',password='123',database='pooldb',charset='utf8' )創建數據庫連接池

?

?

使用數據庫連接池

?

def func():# 檢測當前正在運行連接數的是否小于最大鏈接數，如果不小于則：等待或報raise TooManyConnections異常# 否則# 則優先去初始化時創建的鏈接中獲取鏈接 SteadyDBConnection。# 然后將SteadyDBConnection對象封裝到PooledDedicatedDBConnection中并返回。# 如果最開始創建的鏈接沒有鏈接，則去創建一個SteadyDBConnection對象，再封裝到PooledDedicatedDBConnection中并返回。# 一旦關閉鏈接后，連接就返回到連接池讓后續線程繼續使用。conn = POOL.connection()cursor = conn.cursor()cursor.execute('select * from tb1')result = cursor.fetchall()conn.close()使用數據庫連接池中的鏈接

?

?

自制sqlhelper

?

class MySQLhelper(object):def __init__(self, host, port, dbuser, password, database):self.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=[], # 開始會話前執行的命令列表。如：["set datestyle to ...", "set time zone ..."]ping=0,# ping MySQL服務端，檢查是否服務可用。# 如：0 = None = never, 1 = default = whenever it is requested, 2 = when a cursor is created, 4 = when a query is executed, 7 = alwayshost=host,port=int(port),user=dbuser,password=password,database=database,charset='utf8')def create_conn_cursor(self):conn = self.pool.connection()cursor = conn.cursor(pymysql.cursors.DictCursor)return conn,cursordef fetch_all(self, sql, args):conn,cursor = self.create_conn_cursor()cursor.execute(sql,args)result = cursor.fetchall()cursor.close()conn.close()return resultdef insert_one(self,sql,args):conn,cursor = self.create_conn_cursor()res = cursor.execute(sql,args)conn.commit()print(res)conn.close()return resdef update(self,sql,args):conn,cursor = self.create_conn_cursor()res = cursor.execute(sql,args)conn.commit()print(res)conn.close()return ressqlhelper = MySQLhelper("127.0.0.1", 3306, "root", "1233121234567", "dragon")# sqlhelper.fetch_all("select * from user where id=%s",(1))# sqlhelper.insert_one("insert into user VALUES (%s,%s)",("jinwangba",4))# sqlhelper.update("update user SET name=%s WHERE id=%s",("yinwangba",1)) 好使不好使,試試就知道了

?

轉載于:https://www.cnblogs.com/xiao-xue-di/p/10137629.html

總結

以上是生活随笔為你收集整理的DBUtils - Python数据库连接池的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。