python数据库连接池DBUtils

作者：马育民 • • 阅读：11009

概述

python的数据库连接池较多，这里介绍的是DbUtils

官方文档api：https://cito.github.io/DBUtils/UsersGuide.html
中文翻译：https://blog.csdn.net/zhaihaifei/article/details/53996866

pypi网址：https://pypi.org/project/DBUtils/#description

安装

pip install DBUtils

DbUtils

DBUtils是一套Python数据库连接池包，并允许对非线程安全的数据库接口进行线程安全包装。DBUtils来自Webware for Python。

DBUtils提供两种模式：

  1. PersistentDB
  2. PooledDB

这两种模式的区别，在下面的例子中有详细解释。另外，与数据库驱动也有关系，比如SQLite数据库只能使用PersistentDB

重点掌握 PooledDB

导入模块

3.0.3之前导入模块如下：

from DBUtils.PooledDB  import PooledDB

3.0.3以后导入模块如下：

from dbutils.pooled_db import PooledDB

PooledDB

 # coding=utf-8

from DBUtils.PooledDB  import PooledDB
import mysql.connector as connector

pool = PooledDB(
    creator = connector, #使用链接数据库的模块
    maxconnections = 5,  #连接池允许的最大连接数，0和None表示没有限制
    mincached = 2, #初始化时，连接池至少创建的空闲的连接，0表示不创建
    maxcached = 3, #连接池空闲的最多连接数，0和None表示没有限制
    maxshared = 0, #连接池中最多共享的连接数量，0和None表示全部共享，ps:其实并没有什么用，因为pymsql和MySQLDB等模块中的threadsafety都为1，所有值无论设置多少，_maxcahed永远为0，所以永远是所有链接共享
    blocking = True, #链接池中如果没有可用共享连接后，是否阻塞等待，True表示等待，False表示不等待然后报错
    setsession = [],#开始会话前执行的命令列表
    ping = 0,#ping Mysql 服务端，检查服务是否可用
    host = '127.0.0.1',
    port = 3308,
    user = 'root',
    password = '',
    database = 'test',
    charset = 'utf8'
)
# 获取数据库连接的方式
conn = pool.connection()

例子

 # coding=utf-8

from DBUtils.PooledDB  import PooledDB
import mysql.connector as connector
import dbutil,time
from threading import Timer

conns=[]

def main():
    global conns
    pool = PooledDB(
        creator = connector, #使用链接数据库的模块
        maxconnections = 5,  #连接池允许的最大连接数，0和None表示没有限制
        mincached = 2, #初始化时，连接池至少创建的空闲的连接，0表示不创建
        maxcached = 3, #连接池空闲的最多连接数，0和None表示没有限制
        maxshared = 0, #连接池中最多共享的连接数量，0和None表示全部共享，ps:其实并没有什么用，因为pymsql和MySQLDB等模块中的threadsafety都为1，所有值无论设置多少，_maxcahed永远为0，所以永远是所有链接共享
        blocking = True, #链接池中如果没有可用共享连接后，是否阻塞等待，True表示等待，False表示不等待然后报错
        setsession = [],#开始会话前执行的命令列表
        ping = 0,#ping Mysql 服务端，检查服务是否可用
        host = '127.0.0.1',
        port = 3308,
        user = 'root',
        password = '',
        database = 'test',
        charset = 'utf8'
    )

    Timer(3,close).start()
    db=dbutil.MySQLUtil()

    for item in range(10):
        conn = pool.connection()
        conns.append(conn)
        print(item)

        db.connection=conn
        data=db.queryone("show status like 'Threads_connected'")
        print('数据库总共连接数:',data)
        data=db.queryone("SELECT CONNECTION_ID()")
        print('当前连接的ID:',data)

    time.sleep(50)

def close():
    conns[0].close()
    conns[1].close()

if __name__ == "__main__":
    main()

由于设置了以下参数：

maxconnections = 5,  #连接池允许的最大连接数，0和None表示没有限制
mincached = 2, #初始化时，连接池至少创建的空闲的连接，0表示不创建
blocking = True, #链接池中如果没有可用共享连接后，是否阻塞等待，True表示等待，False表示不等待然后报错

从打印结果可知：

  1. 初始化时，只创建2个数据库连接
  2. 之后从连接池中获取连接
  3. 当连接不够用时，继续创建连接
  4. 当创建5个连接时，就处于等待状态
  5. 当有连接关闭时，并没有真正的关闭，而是回收到连接池中
  6. 可以继续从连接池中获取连接，从连接的ID可知，并没有创建新的，而是使用了回收的连接

PersistentDB

 # coding=utf-8

from DBUtils.PersistentDB  import PersistentDB
import mysql.connector as connector
import dbutil,time
from threading import Timer

pool = PersistentDB(
    creator = connector,  #使用链接数据库的模块
    maxusage = None, #一个链接最多被使用的次数，None表示无限制
    setsession = [], #开始会话前执行的命令
    ping = 0, #ping MySQL服务端,检查服务是否可用
    closeable = False, #conn.close()实际上被忽略，供下次使用，直到线程关闭，自动关闭链接，而等于True时，conn.close()真的被关闭
    threadlocal = None, # 本线程独享值的对象，用于保存链接对象
    host = '127.0.0.1',
    port = 3308,
    user = 'root',
    password = '',
    database = 'test',
    charset = 'utf8'
)
#获取数据库连接
conn = pool.connection()

例子

 # coding=utf-8

from DBUtils.PersistentDB  import PersistentDB
import mysql.connector as connector
import dbutil,time
from threading import Timer

pool = PersistentDB(
    creator = connector,  #使用链接数据库的模块
    maxusage = None, #一个链接最多被使用的次数，None表示无限制
    setsession = [], #开始会话前执行的命令
    ping = 0, #ping MySQL服务端,检查服务是否可用
    closeable = False, #conn.close()实际上被忽略，供下次使用，直到线程关闭，自动关闭链接，而等于True时，conn.close()真的被关闭
    threadlocal = None, # 本线程独享值的对象，用于保存链接对象
    host = '127.0.0.1',
    port = 3308,
    user = 'root',
    password = '',
    database = 'test',
    charset = 'utf8'
)

conns=[]
db=dbutil.MySQLUtil()
for item in range(10):
    conn = pool.connection()
    conns.append(conn)
    print(item)

    db.connection=conn
    # 查询数据库连接的id
    datas=db.queryone("SELECT CONNECTION_ID()")
    print(datas)


time.sleep(50)

从打印结果可知，同一个连接，在反复使用

原文出处：/show_1EF3W4lDlhS0.html

吉ICP备15003043号