Python封装数据库连接池详解_python_

# -*- coding: UTF-8 -*- import pymysql # 数据库信息 DB_TEST_HOST = "127.0.0.1" DB_TEST_PORT = 3308 DB_TEST_DBNAME = "bt" DB_TEST_USER = "root" DB_TEST_PASSWORD = "123456" # 数据库连接编码 DB_CHARSET = "utf8" # mincached : 启动时开启的闲置连接数量(缺省值 0 开始时不创建连接) DB_MIN_CACHED = 5 # maxcached : 连接池中允许的闲置的最多连接数量(缺省值 0 代表不闲置连接池大小) DB_MAX_CACHED = 0 # maxshared : 共享连接数允许的最大数量(缺省值 0 代表所有连接都是专用的)如果达到了最大数量,被请求为共享的连接将会被共享使用 DB_MAX_SHARED = 5 # maxconnecyions : 创建连接池的最大数量(缺省值 0 代表不限制) DB_MAX_CONNECYIONS = 300 # blocking : 设置在连接池达到最大数量时的行为(缺省值 0 或 False 代表返回一个错误 其他代表阻塞直到连接数减少,连接被分配) DB_BLOCKING = True # maxusage : 单个连接的最大允许复用次数(缺省值 0 或 False 代表不限制的复用).当达到最大数时,连接会自动重新连接(关闭和重新打开) DB_MAX_USAGE = 0 # setsession : 一个可选的SQL命令列表用于准备每个会话，如["set datestyle to german", ...] DB_SET_SESSION = None # creator : 使用连接数据库的模块 DB_CREATOR = pymysql

#单例模式函数，用来修饰类 def singleton(cls,*args,**kw): instances = {} def _singleton(): if cls not in instances: instances[cls] = cls(*args,**kw) return instances[cls] return _singleton

from dbutils.pooled_db import PooledDB import db_config as config # import random from singleton import singleton """ @功能：创建数据库连接池 """ class MyConnectionPool(object): # 私有属性 # 能通过对象直接访问，但是可以在本类内部访问； __pool = None # def __init__(self): # self.conn = self.__getConn() # self.cursor = self.conn.cursor() # 创建数据库连接conn和游标cursor def __enter__(self): self.conn = self.__getconn() self.cursor = self.conn.cursor() # 创建数据库连接池 def __getconn(self): if self.__pool is None: # i = random.randint(1, 100) # print("创建线程池的数量"+str(i)) self.__pool = PooledDB( creator=config.DB_CREATOR, mincached=config.DB_MIN_CACHED, maxcached=config.DB_MAX_CACHED, maxshared=config.DB_MAX_SHARED, maxconnections=config.DB_MAX_CONNECYIONS, blocking=config.DB_BLOCKING, maxusage=config.DB_MAX_USAGE, setsession=config.DB_SET_SESSION, host=config.DB_TEST_HOST, port=config.DB_TEST_PORT, user=config.DB_TEST_USER, passwd=config.DB_TEST_PASSWORD, db=config.DB_TEST_DBNAME, use_unicode=False, charset=config.DB_CHARSET ) return self.__pool.connection() # 释放连接池资源 def __exit__(self, exc_type, exc_val, exc_tb): self.cursor.close() self.conn.close() # 关闭连接归还给链接池 # def close(self): # self.cursor.close() # self.conn.close() # 从连接池中取出一个连接 def getconn(self): conn = self.__getconn() cursor = conn.cursor() return cursor, conn # 获取连接池,实例化 @singleton def get_my_connection(): return MyConnectionPool()

import time from db_dbutils_init import get_my_connection """执行语句查询有结果返回结果没有返回0；增/删/改返回变更数据条数，没有返回0""" class MySqLHelper(object): def __init__(self): self.db = get_my_connection() # 从数据池中获取连接 # # def __new__(cls, *args, **kwargs): # if not hasattr(cls, 'inst'): # 单例 # cls.inst = super(MySqLHelper, cls).__new__(cls, *args, **kwargs) # return cls.inst # 封装执行命令 def execute(self, sql, param=None, autoclose=False): """ 【主要判断是否有参数和是否执行完就释放连接】 :param sql: 字符串类型，sql语句 :param param: sql语句中要替换的参数"select %s from tab where id=%s" 其中的%s就是参数 :param autoclose: 是否关闭连接 :return: 返回连接conn和游标cursor """ cursor, conn = self.db.getconn() # 从连接池获取连接 count = 0 try: # count : 为改变的数据条数 if param: count = cursor.execute(sql, param) else: count = cursor.execute(sql) conn.commit() if autoclose: self.close(cursor, conn) except Exception as e: pass return cursor, conn, count # 释放连接 def close(self, cursor, conn): """释放连接归还给连接池""" cursor.close() conn.close() # 查询所有 def selectall(self, sql, param=None): cursor = None conn = None count = None try: cursor, conn, count = self.execute(sql, param) res = cursor.fetchall() return res except Exception as e: print(e) self.close(cursor, conn) return count # 查询单条 def selectone(self, sql, param=None): cursor = None conn = None count = None try: cursor, conn, count = self.execute(sql, param) res = cursor.fetchone() self.close(cursor, conn) return res except Exception as e: print("error_msg:", e.args) self.close(cursor, conn) return count # 增加 def insertone(self, sql, param): cursor = None conn = None count = None try: cursor, conn, count = self.execute(sql, param) # _id = cursor.lastrowid() # 获取当前插入数据的主键id，该id应该为自动生成为好 conn.commit() self.close(cursor, conn) return count except Exception as e: print(e) conn.rollback() self.close(cursor, conn) return count # 增加多行 def insertmany(self, sql, param): """ :param sql: :param param: 必须是元组或列表[(),()]或（（），（）） :return: """ cursor, conn, count = self.db.getconn() try: cursor.executemany(sql, param) conn.commit() return count except Exception as e: print(e) conn.rollback() self.close(cursor, conn) return count # 删除 def delete(self, sql, param=None): cursor = None conn = None count = None try: cursor, conn, count = self.execute(sql, param) self.close(cursor, conn) return count except Exception as e: print(e) conn.rollback() self.close(cursor, conn) return count # 更新 def update(self, sql, param=None): cursor = None conn = None count = None try: cursor, conn, count = self.execute(sql, param) conn.commit() self.close(cursor, conn) return count except Exception as e: print(e) conn.rollback() self.close(cursor, conn) return count # if __name__ == '__main__': # db = MySqLHelper() # sql = "SELECT SLEEP(10)" # db.execute(sql) # time.sleep(20) # TODO 查询单条 # sql1 = 'select * from userinfo where name=%s' # args = 'python' # ret = db.selectone(sql=sql1, param=args) # print(ret) # (None, b'python', b'123456', b'0') # TODO 增加单条 # sql2 = 'insert into hotel_urls(cname,hname,cid,hid,url) values(%s,%s,%s,%s,%s)' # ret = db.insertone(sql2, ('1', '2', '1', '2', '2')) # print(ret) # TODO 增加多条 # sql3 = 'insert into userinfo (name,password) VALUES (%s,%s)' # li = li = [ # ('分省', '123'), # ('到达','456') # ] # ret = db.insertmany(sql3,li) # print(ret) # TODO 删除 # sql4 = 'delete from userinfo WHERE name=%s' # args = 'xxxx' # ret = db.delete(sql4, args) # print(ret) # TODO 更新 # sql5 = r'update userinfo set password=%s WHERE name LIKE %s' # args = ('993333993', '%old%') # ret = db.update(sql5, args) # print(ret)

from dbutils.pooled_db import PooledDB import db_config as config import random from singleton import singleton """ @功能：创建数据库连接池 """ class MyConnectionPool(object): # 私有属性 # 能通过对象直接访问，但是可以在本类内部访问； __pool = None # def __init__(self): # self.conn = self.__getConn() # self.cursor = self.conn.cursor() # 创建数据库连接conn和游标cursor def __enter__(self): self.conn = self.__getconn() self.cursor = self.conn.cursor() # 创建数据库连接池 def __getconn(self): if self.__pool is None: i = random.randint(1, 100) print("线程池的随机数"+str(i)) self.__pool = PooledDB( creator=config.DB_CREATOR, mincached=config.DB_MIN_CACHED, maxcached=config.DB_MAX_CACHED, maxshared=config.DB_MAX_SHARED, maxconnections=config.DB_MAX_CONNECYIONS, blocking=config.DB_BLOCKING, maxusage=config.DB_MAX_USAGE, setsession=config.DB_SET_SESSION, host=config.DB_TEST_HOST, port=config.DB_TEST_PORT, user=config.DB_TEST_USER, passwd=config.DB_TEST_PASSWORD, db=config.DB_TEST_DBNAME, use_unicode=False, charset=config.DB_CHARSET ) return self.__pool.connection() # 释放连接池资源 def __exit__(self, exc_type, exc_val, exc_tb): self.cursor.close() self.conn.close() # 关闭连接归还给链接池 # def close(self): # self.cursor.close() # self.conn.close() # 从连接池中取出一个连接 def getconn(self): conn = self.__getconn() cursor = conn.cursor() return cursor, conn # 获取连接池,实例化 @singleton def get_my_connection(): return MyConnectionPool()

from mysqlhelper import MySqLHelper import time if __name__ == '__main__': sql = "SELECT SLEEP(10)" sql1 = "SELECT SLEEP(15)" db = MySqLHelper() db.execute(sql) db.execute(sql1) time.sleep(20)

show processlist;