Python實現Mysql資料庫連線池
阿新 • • 發佈：2019-01-30
python連線Mysql資料庫：
python程式設計中可以使用MySQLdb進行資料庫的連線及諸如查詢/插入/更新等操作，但是每次連線mysql資料庫請求時，都是獨立的去請求訪問，相當浪費資源，而且訪問數量達到一定數量時，對mysql的效能會產生較大的影響。因此，實際使用中，通常會使用資料庫的連線池技術，來訪問資料庫達到資源複用的目的。
python的資料庫連線池包 DBUtils：
DBUtils是一套Python資料庫連線池包，並允許對非執行緒安全的資料庫介面進行執行緒安全包裝。DBUtils來自Webware for Python。
DBUtils提供兩種外部介面：
- * PersistentDB ：提供執行緒專用的資料庫連線，並自動管理連線。
- * PooledDB ：提供執行緒間可共享的資料庫連線，並自動管理連線。
下載地址：DBUtils 下載解壓後，使用python setup.py install 命令進行安裝
下面利用MySQLdb和DBUtils建立自己的mysql資料庫連線池工具包
在工程目錄下新建package命名為:dbConnecttion，並新建module命名為MySqlConn，下面是MySqlConn.py，該模組建立Mysql的連線池物件，並建立瞭如查詢/插入等通用的操作方法。該部分程式碼實現如下：
# -*- coding: UTF-8 -*- """ Created on 2016年5月7日 @author: baocheng 1、執行帶引數的ＳＱＬ時，請先用sql語句指定需要輸入的條件列表，然後再用tuple/list進行條件批配 ２、在格式ＳＱＬ中不需要使用引號指定資料型別，系統會根據輸入引數自動識別 ３、在輸入的值中不需要使用轉意函式，系統會自動處理 """ import MySQLdb from MySQLdb.cursors import DictCursor from DBUtils.PooledDB import PooledDB #from PooledDB import PooledDB import Config """ Config是一些資料庫的配置檔案 """ class Mysql(object): """ MYSQL資料庫物件，負責產生資料庫連線 , 此類中的連線採用連線池實現獲取連線物件：conn = Mysql.getConn() 釋放連線物件;conn.close()或del conn """ #連線池物件 __pool = None def __init__(self): #資料庫建構函式，從連線池中取出連線，並生成操作遊標 self._conn = Mysql.__getConn() self._cursor = self._conn.cursor() @staticmethod def __getConn(): """ @summary: 靜態方法，從連線池中取出連線 @return MySQLdb.connection """ if Mysql.__pool is None: __pool = PooledDB(creator=MySQLdb, mincached=1 , maxcached=20 , host=Config.DBHOST , port=Config.DBPORT , user=Config.DBUSER , passwd=Config.DBPWD , db=Config.DBNAME,use_unicode=False,charset=Config.DBCHAR,cursorclass=DictCursor) return __pool.connection() def getAll(self,sql,param=None): """ @summary: 執行查詢，並取出所有結果集 @param sql:查詢ＳＱＬ，如果有查詢條件，請只指定條件列表，並將條件值使用引數[param]傳遞進來 @param param: 可選引數，條件列表值（元組/列表） @return: result list(字典物件)/boolean 查詢到的結果集 """ if param is None: count = self._cursor.execute(sql) else: count = self._cursor.execute(sql,param) if count>0: result = self._cursor.fetchall() else: result = False return result def getOne(self,sql,param=None): """ @summary: 執行查詢，並取出第一條 @param sql:查詢ＳＱＬ，如果有查詢條件，請只指定條件列表，並將條件值使用引數[param]傳遞進來 @param param: 可選引數，條件列表值（元組/列表） @return: result list/boolean 查詢到的結果集 """ if param is None: count = self._cursor.execute(sql) else: count = self._cursor.execute(sql,param) if count>0: result = self._cursor.fetchone() else: result = False return result def getMany(self,sql,num,param=None): """ @summary: 執行查詢，並取出num條結果 @param sql:查詢ＳＱＬ，如果有查詢條件，請只指定條件列表，並將條件值使用引數[param]傳遞進來 @param num:取得的結果條數 @param param: 可選引數，條件列表值（元組/列表） @return: result list/boolean 查詢到的結果集 """ if param is None: count = self._cursor.execute(sql) else: count = self._cursor.execute(sql,param) if count>0: result = self._cursor.fetchmany(num) else: result = False return result def insertOne(self,sql,value): """ @summary: 向資料表插入一條記錄 @param sql:要插入的ＳＱＬ格式 @param value:要插入的記錄資料tuple/list @return: insertId 受影響的行數 """ self._cursor.execute(sql,value) return self.__getInsertId() def insertMany(self,sql,values): """ @summary: 向資料表插入多條記錄 @param sql:要插入的ＳＱＬ格式 @param values:要插入的記錄資料tuple(tuple)/list[list] @return: count 受影響的行數 """ count = self._cursor.executemany(sql,values) return count def __getInsertId(self): """ 獲取當前連線最後一次插入操作生成的id,如果沒有則為０ """ self._cursor.execute("SELECT @@IDENTITY AS id") result = self._cursor.fetchall() return result[0]['id'] def __query(self,sql,param=None): if param is None: count = self._cursor.execute(sql) else: count = self._cursor.execute(sql,param) return count def update(self,sql,param=None): """ @summary: 更新資料表記錄 @param sql: ＳＱＬ格式及條件，使用(%s,%s) @param param: 要更新的 值 tuple/list @return: count 受影響的行數 """ return self.__query(sql,param) def delete(self,sql,param=None): """ @summary: 刪除資料表記錄 @param sql: ＳＱＬ格式及條件，使用(%s,%s) @param param: 要刪除的條件 值 tuple/list @return: count 受影響的行數 """ return self.__query(sql,param) def begin(self): """ @summary: 開啟事務 """ self._conn.autocommit(0) def end(self,option='commit'): """ @summary: 結束事務 """ if option=='commit': self._conn.commit() else: self._conn.rollback() def dispose(self,isEnd=1): """ @summary: 釋放連線池資源 """ if isEnd==1: self.end('commit') else: self.end('rollback'); self._cursor.close() self._conn.close()
配置檔案模組Cnofig,包括資料庫的連線資訊/使用者名稱密碼等：
#coding:utf-8
'''
Created on 2016年5月7日
@author: baocheng
'''
DBHOST = "localhost"
DBPORT = 33606
DBUSER = "zbc"
DBPWD = "123456"
DBNAME = "test"
DBCHAR = "utf8"
當然，還有很多其他引數可以配置：#coding:utf-8 ''' @author: baocheng ''' from MySqlConn import Mysql from _sqlite3 import Row #申請資源 mysql = Mysql() sqlAll = "SELECT tb.uid as uid, group_concat(tb.goodsname) as goodsname FROM ( SELECT goods.uid AS uid, IF ( ISNULL(goodsrelation.goodsname), goods.goodsID, goodsrelation.goodsname ) AS goodsname FROM goods LEFT JOIN goodsrelation ON goods.goodsID = goodsrelation.goodsId ) tb GROUP BY tb.uid" result = mysql.getAll(sqlAll) if result : print "get all" for row in result : print "%s\t%s"%(row["uid"],row["goodsname"]) sqlAll = "SELECT tb.uid as uid, group_concat(tb.goodsname) as goodsname FROM ( SELECT goods.uid AS uid, IF ( ISNULL(goodsrelation.goodsname), goods.goodsID, goodsrelation.goodsname ) AS goodsname FROM goods LEFT JOIN goodsrelation ON goods.goodsID = goodsrelation.goodsId ) tb GROUP BY tb.uid" result = mysql.getMany(sqlAll,2) if result : print "get many" for row in result : print "%s\t%s"%(row["uid"],row["goodsname"]) result = mysql.getOne(sqlAll) print "get one" print "%s\t%s"%(result["uid"],result["goodsname"]) #釋放資源 mysql.dispose()
- dbapi ：資料庫介面
- mincached ：啟動時開啟的空連線數量
- maxcached ：連線池最大可用連線數量
- maxshared ：連線池最大可共享連線數量
- maxconnections ：最大允許連線數量
- blocking ：達到最大數量時是否阻塞
- maxusage ：單個連線最大複用次數
至此，python中的mysql連線池實現完了，下次就直接拿來用就好了。
部落格地址：data mining club