每篇前言：
????作者介紹：【孤寒者】—CSDN全棧領(lǐng)域優(yōu)質(zhì)創(chuàng)作者、HDZ核心組成員、華為云享專(zhuān)家Python全棧領(lǐng)域博主、CSDN原力計(jì)劃作者
- ????本文已收錄于Flask框架從入門(mén)到實(shí)戰(zhàn)專(zhuān)欄：《Flask框架從入門(mén)到實(shí)戰(zhàn)》
- ????熱門(mén)專(zhuān)欄推薦：《Python全棧系列教程》、《爬蟲(chóng)從入門(mén)到精通系列教程》、《爬蟲(chóng)進(jìn)階+實(shí)戰(zhàn)系列教程》、《Scrapy框架從入門(mén)到實(shí)戰(zhàn)》、《Flask框架從入門(mén)到實(shí)戰(zhàn)》、《Django框架從入門(mén)到實(shí)戰(zhàn)》、《Tornado框架從入門(mén)到實(shí)戰(zhàn)》、《前端系列教程》。
- ?????本專(zhuān)欄面向廣大程序猿，為的是大家都做到Python全棧技術(shù)從入門(mén)到精通，穿插有很多實(shí)戰(zhàn)優(yōu)化點(diǎn)。
- ????訂閱專(zhuān)欄后可私聊進(jìn)一千多人Python全棧交流群（手把手教學(xué)，問(wèn)題解答）； 進(jìn)群可領(lǐng)取Python全棧教程視頻 + 多得數(shù)不過(guò)來(lái)的計(jì)算機(jī)書(shū)籍：基礎(chǔ)、Web、爬蟲(chóng)、數(shù)據(jù)分析、可視化、機(jī)器學(xué)習(xí)、深度學(xué)習(xí)、人工智能、算法、面試題等。
- ????加入我一起學(xué)習(xí)進(jìn)步，一個(gè)人可以走的很快，一群人才能走的更遠(yuǎn)！
引子：
在上一個(gè)demo項(xiàng)目中，登錄部分驗(yàn)證是直接寫(xiě)死的，本文模擬實(shí)際生產(chǎn)，查詢(xún)MySQL數(shù)據(jù)庫(kù)做驗(yàn)證。
一個(gè)很low的方法是：
項(xiàng)目根目錄下創(chuàng)建utils/sql.py：
import pymysql
class SQLHelper(object):
@staticmethod
def open():
conn = pymysql.connect(host='127.0.0.1', port=3306, password='123456', db='UserInfo')
cursor = conn.cursor(cursor=pymysql.cursors.DictCursor)
return conn, cursor
@staticmethod
def close(conn, cursor):
conn.commit()
cursor.close()
conn.close()
@classmethod
def fetch_one(cls, sql, args):
conn, cursor = cls.open()
cursor.execute(sql, args)
obj = cursor.fetchone()
cls.close(conn, cursor)
return obj
上面這種用法很?chē)?yán)重且明顯的一個(gè)問(wèn)題是： 每次登錄一次都要和數(shù)據(jù)庫(kù)創(chuàng)建一個(gè)連接！
解決方法就是使用DBUtils三方庫(kù)：
DBUtils庫(kù)
pip install DBUtils==1.3
DBUtils 是一套用于管理數(shù)據(jù)庫(kù)連接池的Python包，為高頻度高并發(fā)的數(shù)據(jù)庫(kù)訪問(wèn)提供更好的性能，可以自動(dòng)管理連接對(duì)象的創(chuàng)建和釋放。并允許對(duì)非線(xiàn)程安全的數(shù)據(jù)庫(kù)接口進(jìn)行線(xiàn)程安全包裝。
這種連接池有兩種連接模式：
-
PersistentDB：提供線(xiàn)程專(zhuān)用的數(shù)據(jù)庫(kù)連接，并自動(dòng)管理連接。
為每個(gè)線(xiàn)程創(chuàng)建一個(gè)連接，線(xiàn)程即使調(diào)用了close方法，也不會(huì)關(guān)閉，只是把連接重新放到連接池，供自己線(xiàn)程再次使用。當(dāng)線(xiàn)程終止時(shí)，連接自動(dòng)關(guān)閉！
-
PooledDB：提供線(xiàn)程間可共享的數(shù)據(jù)庫(kù)連接，并自動(dòng)管理連接。
創(chuàng)建一批連接到連接池，供所有線(xiàn)程共享使用。
PS：由于pymysql的threadsafety值為1，而DBUtils庫(kù)用的內(nèi)部又是用的pymysql，所以該模式連接池中的線(xiàn)程會(huì)被所有線(xiàn)程共享。
實(shí)測(cè)證明 PersistentDB 的速度是最高的（即第一種模式），但是在某些特殊情況下，數(shù)據(jù)庫(kù)的連接過(guò)程可能異常緩慢，而此時(shí)的PooledDB（即模式二，所以推薦這個(gè)）則可以提供相對(duì)來(lái)說(shuō)平均連接時(shí)間比較短的管理方式。
模式一（底層使用threading.local實(shí)現(xiàn)）：
- 了解即可~~~
from DBUtils.PersistentDB import PersistentDB
import pymysql
POOL = PersistentDB(
creator=pymysql, # 連接數(shù)據(jù)庫(kù)的模塊
maxusage=None, # 一個(gè)數(shù)據(jù)庫(kù)連接最多被重復(fù)使用的次數(shù)，None表示無(wú)限制
setsession=[], # 開(kāi)始會(huì)話(huà)前執(zhí)行的命令列表。如["set time zone ..."]
ping=0, # ping MySQL客戶(hù)端，檢查服務(wù)是否可用。 【一般用0,4,7】
# 0 = None = never; 1 = default = whenever if is requested; 2 = when a cursor is created; 4 = when a query is executed; 7 = always
closeable=False, # False：conn.close()實(shí)際上被忽略，供下次使用，在線(xiàn)程關(guān)閉時(shí)，才會(huì)自動(dòng)關(guān)閉連接；True：conn.close()則關(guān)閉連接，再次調(diào)用就是一個(gè)新的連接了
threadlocal=None, # 本線(xiàn)程獨(dú)享值的對(duì)象，用于保存鏈接對(duì)象，如果鏈接對(duì)象被重置，也清除
host='127.0.0.1',
port=6379,
user='root',
password='123456',
database='UserInfo',
charset='utf8'
)
使用：
def demo():
conn = POOL.connection(shareable=False)
cursor = conn.cursor()
cursor.execute('select * from users')
result = cursor.fetchall()
cursor.close()
conn.close()
模式二：
- 用這個(gè)~~~
from DBUtils.PooledDB import PooledDB
import pymysql
POOL = PooledDB(
creator=pymysql, # 連接數(shù)據(jù)庫(kù)的模塊
maxconnections=6, # 連接池允許的最大連接數(shù)， 0和None表示不限制
mincached=2, # 初始化時(shí)，連接池中至少創(chuàng)建的空閑的連接，0表示不創(chuàng)建
maxcached=5, # 連接池中最多閑置的連接，0和None不限制
maxshared=3, # 連接池中最多共享的連接數(shù)量，0和None表示全部共享【默認(rèn)為0，而且哪怕設(shè)置別的值也無(wú)用?。?！下面會(huì)將為啥】
blocking=True, # 連接池中如果沒(méi)有可用連接后，是否阻塞等待。True：等待；False：不等待直接報(bào)錯(cuò)
maxusage=None, # 一個(gè)連接最多被重復(fù)使用的次數(shù)，None表示無(wú)限制
setsession=[], # 開(kāi)始會(huì)話(huà)前執(zhí)行的命令列表。如["set time zone ..."]
ping=0, # ping MySQL客戶(hù)端，檢查服務(wù)是否可用。 【一般用0,4,7】
# 0 = None = never; 1 = default = whenever if is requested; 2 = when a cursor is created; 4 = when a query is executed; 7 = always
host='127.0.0.1',
port=6379,
user='root',
password='123456',
database='UserInfo',
charset='utf8'
)
-
為什么設(shè)置maxshared參數(shù)的值是無(wú)用的！因?yàn)镈BUtils使用的pymysql，而pymysql內(nèi)部threadsafety值為1?？丛创a：
import pymysql
DBUtils源碼：
from DBUtils.PooledDB import PooledDB
使用：
def demo():
"""
檢測(cè)當(dāng)前正在運(yùn)行連接數(shù)是否小于最大連接數(shù)；如果不小于則等待或報(bào)錯(cuò)：raise TooManyConnection異常
否則則優(yōu)先去初始化時(shí)創(chuàng)建的連接中獲取連接：SteadyDBConnection
然后將SteadyDBConnection對(duì)象封裝到PooledDedicatedDBConnection中并返回。
如果最開(kāi)始創(chuàng)建的連接沒(méi)有連接，則去創(chuàng)建一個(gè)SteadyDBConnection對(duì)象，再封裝到PooledDedicatedDBConnection中并返回。
一旦關(guān)閉連接后，連接就返回到連接池讓后續(xù)線(xiàn)程繼續(xù)使用~
"""
conn = POOL.connection()
cursor = conn.cursor()
cursor.execute('select * from users')
result = cursor.fetchall()
conn.close()
Flask中使用
方式一：直接將DBUtils初始化放到settings.py文件中
文章來(lái)源：http://www.zghlxwxcb.cn/news/detail-849901.html
方式二：從utils文件夾中導(dǎo)入
文章來(lái)源地址http://www.zghlxwxcb.cn/news/detail-849901.html
腳本使用DBUtils代碼demo：
# coding=utf-8
"""
使用DBUtils數(shù)據(jù)庫(kù)連接池中的連接，操作數(shù)據(jù)庫(kù)
"""
import datetime
import pymysql
from DBUtils.PooledDB import PooledDB
class MysqlClient(object):
def __init__(self, **kwargs):
self.pool = self.create_pool(**kwargs)
self.connection = None
self.cursor = None
def create_pool(self, **kwargs):
return PooledDB(
pymysql,
mincached=kwargs.get('mincached', 10),
maxcached=kwargs.get('maxcached', 20),
maxshared=kwargs.get('maxshared', 10),
maxconnections=kwargs.get('maxconnections', 200),
blocking=kwargs.get('blocking', True),
maxusage=kwargs.get('maxusage', 100),
setsession=kwargs.get('setsession', None),
reset=kwargs.get('reset', True),
host=kwargs.get('host', '127.0.0.1'),
port=kwargs.get('port', 3306),
db=kwargs.get('db', 'mysqldemo'),
user=kwargs.get('user', 'root'),
passwd=kwargs.get('passwd', '123456'),
charset=kwargs.get('charset', 'utf8mb4'),
cursorclass=pymysql.cursors.DictCursor
)
def get_conn_cursor(self):
self.connection = self.pool.connection()
self.cursor = self.connection.cursor()
def close(self):
try:
if self.cursor:
self.cursor.close()
if self.connection:
self.connection.close()
except Exception as e:
print(e)
def execute(self, sql, param=()):
try:
self.get_conn_cursor()
count = self.cursor.execute(sql, param)
print(count)
return count
finally:
self.close()
def __dict_datetime_obj_to_str(self, result_dict):
"""把字典里面的datetime對(duì)象轉(zhuǎn)成字符串，使json轉(zhuǎn)換不出錯(cuò)"""
if result_dict:
return {k: v.__str__() if isinstance(v, datetime.datetime) else v for k, v in result_dict.items()}
return result_dict
def select_one(self, sql, param=()):
"""查詢(xún)單個(gè)結(jié)果"""
try:
self.get_conn_cursor()
count = self.execute(sql, param)
result = self.cursor.fetchone()
result = self.__dict_datetime_obj_to_str(result)
return count, result
finally:
self.close()
def select_many(self, sql, param=()):
"""查詢(xún)多個(gè)結(jié)果"""
try:
self.get_conn_cursor()
count = self.execute(sql, param)
result = self.cursor.fetchall()
result = [self.__dict_datetime_obj_to_str(row_dict) for row_dict in result]
return count, result
finally:
self.close()
def begin(self):
"""開(kāi)啟事務(wù)"""
try:
self.get_conn_cursor()
self.connection.autocommit(False)
except Exception as e:
print(e)
def end(self, option='commit'):
"""結(jié)束事務(wù)"""
try:
if option == 'commit':
self.connection.commit()
else:
self.connection.rollback()
except Exception as e:
print(e)
finally:
self.connection.autocommit(True)
if __name__ == "__main__":
mc = MysqlClient()
sql1 = 'SELECT * FROM customers WHERE customerNumber = 103'
result1 = mc.select_one(sql1)
print(result1[1])
sql2 = 'SELECT * FROM customers WHERE customerNumber IN (%s,%s,%s)'
param = (103, 144, 145)
print(mc.select_many(sql2, param)[1])
到了這里，關(guān)于（二十七）Flask之?dāng)?shù)據(jù)庫(kù)連接池DBUtils庫(kù)的文章就介紹完了。如果您還想了解更多內(nèi)容，請(qǐng)?jiān)谟疑辖撬阉鱐OY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！
