一、问题引入
如果是简单的数据库操作，使用pymysql这样就可以解决问题了：
import pymysql conn = pymysql.connect( host='localhost', port=3306, user='xxx', password='xxx', database='test', charset='utf8' ) cursor = conn.cursor() cursor.execute("select * from teacher") result = cursor.fetchall() cursor.close() conn.close()
这样可以简单的完成一次查询，但是往往为了效率，会使用多进程或者多线程的方式，比如同时向数据库插入几百条或者查询多个结果，这样就会造成同时创建过多的连接，而数据库的最大连接数十有限制的，默认为200：
mysql> show global variables like 'max_connections'; +-----------------+-------+ | Variable_name | Value | +-----------------+-------+ | max_connections | 200 | +-----------------+-------+ 1 row in set (0.00 sec)
为了解决这个问题，可以引入数据库连接池。DBUtils可以很好的解决这个问题。
二、DBUtils
DBUtils是一套Python数据库连接池包，可为高并发访问数据库提供更好的性能服务，并允许对非线程安全的数据库接口进行线程安全包装。
它有两种模式，分别为：
- PersistentDB
- PooledDB
（一）PersistentDB
为每一个线程创建独有的连接，即使调用close方法也不会关闭，只是把连接放入到连接池中供后续线程继续使用，当线程结束后，连接也就会自动关闭。
from DBUtils.PersistentDB import PersistentDB import pymysql pool = PersistentDB( pymysql, #数据库连接模块 maxusage=None, # 一个链接最多被重复使用的次数，None表示无限制 setsession=[], # 开始会话前执行的命令列表 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 = always 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' )
使用时直接调用连接池对象的connection方法获取连接即可。
def index(): conn = pool.connection(shareable=False) cursor = conn.cursor() cursor.execute('select * from table') result = cursor.fetchall() print(result) cursor.close() conn.close()
（二）PooledDB
为所有的线程创建共享的连接，创建一批连接到连接池中，供所有的线程使用。
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表示全部共享。 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 = always host='127.0.0.1', port=3306, user='root', password='123', database='test', charset='utf8' )
使用时直接调用连接池对象的connection方法获取连接即可。
def index(): conn = pool.connection() cursor = conn.cursor() cursor.execute('select * from table') result = cursor.fetchall() conn.close()
pip install DBUtils==1.3
三、使用实例
（一）方法一
1、创建连接池实例
import pymysql from DBUtils.PooledDB import PooledDB pool = PooledDB( creator=pymysql, maxconnections=6, mincached=2, maxcached=5, maxshared=3, blocking=True, maxusage=None, setsession=[], ping=0, host='127.0.0.1', port=3306, user='root', password='123', database='test', charset='utf8' )
2、使用实例
上面已经创建好了连接池的实例，现在只需要使用这个实例即可。
class DBManger: @staticmethod def fetch_one(sql,args=None): conn = pool.connection() cursor = conn.cursor() if not args: cursor.execute(sql) cursor.execute(sql,args) result = cursor.fetchone() cursor.close() conn.close() return result
如果调用它，只需要进行如下操作即可：
def index(): sql = 'select * from teacher where tid>%s' parames = 1 result = DBManger.fetch_one(sql,(parames,)) #注意此处传参，可能传入的是字符串，也有可能是列表等（sql语句中含有in关键字） print(result)
（二）方法二
使用封装的方式进行调用：
import pymysql from DBUtils.PooledDB import PooledDB class DBManger: __pool = None def __init__(self, creator=pymysql, maxconnections=200,mincached=10,maxcached=20,maxshared=3, blocking=True, maxusage=None,setsession=None,host='127.0.0.1',port=3306, user='root',password='123',database='test',charset='utf8' ): """ :param creator: 数据库连接模块 :param maxconnections: 连接池允许的最大连接数，0和None表示不限制连接数 :param mincached:连接池中空闲连接的初始数量 :param maxcached:连接池中空闲连接的最大数量 :param maxshared:共享连接的最大数量 :param blocking:超过最大连接数量时候的表现，为True等待连接数量下降，为false直接报错处理 :param maxusage:单个连接的最大重复使用次数 :param setsession:开始会话前执行的命令列表。如：["set datestyle to ...", "set time zone ..."] :param ping: ping MySQL服务端，检查是否服务可用。# 如：0 = None = never :param host:主机 :param port:端口号 :param user:用户名 :param password:密码 :param database:数据库 :param charset:编码 """ if not self.__pool: self.__pool = PooledDB( creator=creator, maxconnections=maxconnections, mincached=mincached, maxcached=maxcached, maxshared=maxshared, blocking=blocking, maxusage=maxusage, setsession=setsession, host=host, port=port, user=user, password=password, database=database, charset=charset, cursorclass=pymysql.cursors.DictCursor #返回字典形式的列表 ) self._conn = None self._cursor = None self.__get_conn() def __get_conn(self): self._conn = self.__pool.connection() self._cursor = self._conn.cursor() def close(self): """释放连接给连接池""" self._cursor.close() self._conn.close() def get_one(self,sql,args=None): """ 获取一条数据 :param sql: :param args: :return: """ try: if not args: self._cursor.excute(sql) print(sql,args) self._cursor.execute(sql,args) result = self._cursor.fetchone() self.close() return result except Exception as e: self.close() def get_many(self,sql,args=None): """ 获取多条数据 :param sql: :param args: :return: """ try: if not args: self._cursor.excute(sql) self._cursor.execute(sql,args) result = self._cursor.fetchall() self.close() return result except Exception as e: self.close() def insert_one(self,sql,args=None): """ 插入单条数据 :param sql: :param args: :return: """ try: if not args: self._cursor.excute(sql) count = self._cursor.execute(sql, args) self._conn.commit() self.close() return count except Exception as e: self._conn.rollback() self.close() def insert_many(self,sql,args=None): """ 插入多条数据 :param sql: :param args: :return: """ try: if not args: self._cursor.executemany(sql) print(args) count = self._cursor.executemany(sql, args) self._conn.commit() self.close() return count except Exception as e: self._conn.rollback() self.close() def delete(self,sql,args=None): """ 删除数据 :param sql: :param args: :return: """ try: if not args: self._cursor.execute(sql) count = self._cursor.execute(sql, args) self._conn.commit() self.close() return count except Exception as e: self._conn.rollback() self.close() def update(self,sql,args=None): """ 更新数据 :param sql: :param args: :return: """ try: if not args: self._cursor.execute(sql) count = self._cursor.execute(sql, args) self._conn.commit() self.close() return count except Exception as e: self._conn.rollback() self.close()
然后就可以进行调用了。
def index(): db = DBManger() #查询一条数据 sql = 'select * from teacher where tid>%s' param = 1 result1 = db.get_one(sql,(param,)) print(result1) #{'tid': 2, 'tname': '王五'} #查询多条语句 sql = 'select * from teacher where tid>%s' param = 1 result2 = db.get_many(sql,(param,)) print(result2) #[{'tid': 2, 'tname': '王五'}, {'tid': 3, 'tname': '赵六'}, {'tid': 4, 'tname': '张三'}] #插入一条数据 sql = 'insert into teacher(tid,tname) values (%s,%s) ' param = ((12,'李田')) #1 result1 = db.insert_one(sql,param) print(result1) #{'tid': 2, 'tname': '王五'} #插入多条数据s sql = 'insert into teacher(tid,tname) values (%s,%s) ' param = ((9,'刘大'),(10,'刘二')) result1 = db.insert_many(sql,param) print(result1) #2 #删除数据 sql = 'delete from teacher where tid>%s' param = (4,) result1 = db.delete(sql,param) #4 删除4条符合条件的数据 print(result1) #2 #更新数据 sql = 'update teacher set tname=%s where tid>%s' param = ('kk',2,) result1 = db.update(sql,param) #2 将符合条件的两条语句进行了修改 print(result1) if __name__ == '__main__': index()