python3 实现mysql数据库连接池的示例代码

向2824

dbutils封装文件传送门
DBUtils是一套Python数据库连接池包，并允许对非线程安全的数据库接口进行线程安全包装。DBUtils来自Webware for Python。
DBUtils提供两种外部接口：

   

• PersistentDB ：提供线程专用的数据库连接，并自动管理连接。   
  
• PooledDB ：提供线程间可共享的数据库连接，并自动管理连接。
  

需要库
1、DBUtils pip install DBUtils
2、pymysql pip install pymysql/MySQLdb
创建DButils组件
db_config.py 配置文件

1. # -*- coding: UTF-8 -*-
   
2. import pymysql
   
3. 
   
4. # 数据库信息
   
5. DB_TEST_HOST = "127.0.0.1"
   
6. DB_TEST_PORT = 3306
   
7. DB_TEST_DBNAME = "ball"
   
8. DB_TEST_USER = "root"
   
9. DB_TEST_PASSWORD = "123456"
   
10. 
    
11. # 数据库连接编码
    
12. DB_CHARSET = "utf8"
    
13. 
    
14. # mincached : 启动时开启的闲置连接数量(缺省值 0 开始时不创建连接)
    
15. DB_MIN_CACHED = 10
    
16. 
    
17. # maxcached : 连接池中允许的闲置的最多连接数量(缺省值 0 代表不闲置连接池大小)
    
18. DB_MAX_CACHED = 10
    
19. 
    
20. # maxshared : 共享连接数允许的最大数量(缺省值 0 代表所有连接都是专用的)如果达到了最大数量,被请求为共享的连接将会被共享使用
    
21. DB_MAX_SHARED = 20
    
22. 
    
23. # maxconnecyions : 创建连接池的最大数量(缺省值 0 代表不限制)
    
24. DB_MAX_CONNECYIONS = 100
    
25. 
    
26. # blocking : 设置在连接池达到最大数量时的行为(缺省值 0 或 False 代表返回一个错误<tomany......> 其他代表阻塞直到连接数减少,连接被分配)
    
27. DB_BLOCKING = True
    
28. 
    
29. # maxusage : 单个连接的最大允许复用次数(缺省值 0 或 False 代表不限制的复用).当达到最大数时,连接会自动重新连接(关闭和重新打开)
    
30. DB_MAX_USAGE = 0
    
31. 
    
32. # setsession : 一个可选的SQL命令列表用于准备每个会话，如["set datestyle to german", ...]
    
33. DB_SET_SESSION = None
    
34. 
    
35. # creator : 使用连接数据库的模块
    
36. DB_CREATOR = pymysql

复制代码

db_dbutils_init.py 创建数据池初始化

1. from DBUtils.PooledDB import PooledDB
   
2. import db_config as config
   
3. 
   
4. """
   
5. @功能：创建数据库连接池
   
6. """
   
7. 
   
8. 
   
9. class MyConnectionPool(object):
   
10.     __pool = None
    
11. 
    
12.     # def __init__(self):
    
13.     #     self.conn = self.__getConn()
    
14.     #     self.cursor = self.conn.cursor()
    
15. 
    
16.     # 创建数据库连接conn和游标cursor
    
17.     def __enter__(self):
    
18.         self.conn = self.__getconn()
    
19.         self.cursor = self.conn.cursor()
    
20. 
    
21.     # 创建数据库连接池
    
22.     def __getconn(self):
    
23.         if self.__pool is None:
    
24.             self.__pool = PooledDB(
    
25.                 creator=config.DB_CREATOR,
    
26.                 mincached=config.DB_MIN_CACHED,
    
27.                 maxcached=config.DB_MAX_CACHED,
    
28.                 maxshared=config.DB_MAX_SHARED,
    
29.                 maxconnections=config.DB_MAX_CONNECYIONS,
    
30.                 blocking=config.DB_BLOCKING,
    
31.                 maxusage=config.DB_MAX_USAGE,
    
32.                 setsession=config.DB_SET_SESSION,
    
33.                 host=config.DB_TEST_HOST,
    
34.                 port=config.DB_TEST_PORT,
    
35.                 user=config.DB_TEST_USER,
    
36.                 passwd=config.DB_TEST_PASSWORD,
    
37.                 db=config.DB_TEST_DBNAME,
    
38.                 use_unicode=False,
    
39.                 charset=config.DB_CHARSET
    
40.             )
    
41.         return self.__pool.connection()
    
42. 
    
43.     # 释放连接池资源
    
44.     def __exit__(self, exc_type, exc_val, exc_tb):
    
45.         self.cursor.close()
    
46.         self.conn.close()
    
47. 
    
48.     # 关闭连接归还给链接池
    
49.     # def close(self):
    
50.     #     self.cursor.close()
    
51.     #     self.conn.close()
    
52. 
    
53.     # 从连接池中取出一个连接
    
54.     def getconn(self):
    
55.         conn = self.__getconn()
    
56.         cursor = conn.cursor()
    
57.         return cursor, conn
    
58. 
    
59. 
    
60. # 获取连接池,实例化
    
61. def get_my_connection():
    
62.     return MyConnectionPool()

复制代码

制作mysqlhelper.py

1. from db_dbutils_init import get_my_connection
   
2. 
   
3. """执行语句查询有结果返回结果没有返回0；增/删/改返回变更数据条数，没有返回0"""
   
4. 
   
5. 
   
6. class MySqLHelper(object):
   
7.     def __init__(self):
   
8.         self.db = get_my_connection()  # 从数据池中获取连接
   
9. 
   
10.     def __new__(cls, *args, **kwargs):
    
11.         if not hasattr(cls, 'inst'):  # 单例
    
12.             cls.inst = super(MySqLHelper, cls).__new__(cls, *args, **kwargs)
    
13.         return cls.inst
    
14. 
    
15.     # 封装执行命令
    
16.     def execute(self, sql, param=None, autoclose=False):
    
17.         """
    
18.         【主要判断是否有参数和是否执行完就释放连接】
    
19.         :param sql: 字符串类型，sql语句
    
20.         :param param: sql语句中要替换的参数"select %s from tab where id=%s" 其中的%s就是参数
    
21.         :param autoclose: 是否关闭连接
    
22.         :return: 返回连接conn和游标cursor
    
23.         """
    
24.         cursor, conn = self.db.getconn()  # 从连接池获取连接
    
25.         count = 0
    
26.         try:
    
27.             # count : 为改变的数据条数
    
28.             if param:
    
29.                 count = cursor.execute(sql, param)
    
30.             else:
    
31.                 count = cursor.execute(sql)
    
32.             conn.commit()
    
33.             if autoclose:
    
34.                 self.close(cursor, conn)
    
35.         except Exception as e:
    
36.             pass
    
37.         return cursor, conn, count
    
38. 
    
39.     # 执行多条命令
    
40.     # def executemany(self, lis):
    
41.     #     """
    
42.     #     :param lis: 是一个列表，里面放的是每个sql的字典'[{"sql":"xxx","param":"xx"}....]'
    
43.     #     :return:
    
44.     #     """
    
45.     #     cursor, conn = self.db.getconn()
    
46.     #     try:
    
47.     #         for order in lis:
    
48.     #             sql = order['sql']
    
49.     #             param = order['param']
    
50.     #             if param:
    
51.     #                 cursor.execute(sql, param)
    
52.     #             else:
    
53.     #                 cursor.execute(sql)
    
54.     #         conn.commit()
    
55.     #         self.close(cursor, conn)
    
56.     #         return True
    
57.     #     except Exception as e:
    
58.     #         print(e)
    
59.     #         conn.rollback()
    
60.     #         self.close(cursor, conn)
    
61.     #         return False
    
62. 
    
63.     # 释放连接
    
64.     def close(self, cursor, conn):
    
65.         """释放连接归还给连接池"""
    
66.         cursor.close()
    
67.         conn.close()
    
68. 
    
69.     # 查询所有
    
70.     def selectall(self, sql, param=None):
    
71.         try:
    
72.             cursor, conn, count = self.execute(sql, param)
    
73.             res = cursor.fetchall()
    
74.             return res
    
75.         except Exception as e:
    
76.             print(e)
    
77.             self.close(cursor, conn)
    
78.             return count
    
79. 
    
80.     # 查询单条
    
81.     def selectone(self, sql, param=None):
    
82.         try:
    
83.             cursor, conn, count = self.execute(sql, param)
    
84.             res = cursor.fetchone()
    
85.             self.close(cursor, conn)
    
86.             return res
    
87.         except Exception as e:
    
88.             print("error_msg:", e.args)
    
89.             self.close(cursor, conn)
    
90.             return count
    
91. 
    
92.     # 增加
    
93.     def insertone(self, sql, param):
    
94.         try:
    
95.             cursor, conn, count = self.execute(sql, param)
    
96.             # _id = cursor.lastrowid()  # 获取当前插入数据的主键id，该id应该为自动生成为好
    
97.             conn.commit()
    
98.             self.close(cursor, conn)
    
99.             return count
    
100.             # 防止表中没有id返回0
     
101.             # if _id == 0:
     
102.             #     return True
     
103.             # return _id
     
104.         except Exception as e:
     
105.             print(e)
     
106.             conn.rollback()
     
107.             self.close(cursor, conn)
     
108.             return count
     
109. 
     
110.     # 增加多行
     
111.     def insertmany(self, sql, param):
     
112.         """
     
113.         :param sql:
     
114.         :param param: 必须是元组或列表[(),()]或（（），（））
     
115.         :return:
     
116.         """
     
117.         cursor, conn, count = self.db.getconn()
     
118.         try:
     
119.             cursor.executemany(sql, param)
     
120.             conn.commit()
     
121.             return count
     
122.         except Exception as e:
     
123.             print(e)
     
124.             conn.rollback()
     
125.             self.close(cursor, conn)
     
126.             return count
     
127. 
     
128.     # 删除
     
129.     def delete(self, sql, param=None):
     
130.         try:
     
131.             cursor, conn, count = self.execute(sql, param)
     
132.             self.close(cursor, conn)
     
133.             return count
     
134.         except Exception as e:
     
135.             print(e)
     
136.             conn.rollback()
     
137.             self.close(cursor, conn)
     
138.             return count
     
139. 
     
140.     # 更新
     
141.     def update(self, sql, param=None):
     
142.         try:
     
143.             cursor, conn, count = self.execute(sql, param)
     
144.             conn.commit()
     
145.             self.close(cursor, conn)
     
146.             return count
     
147.         except Exception as e:
     
148.             print(e)
     
149.             conn.rollback()
     
150.             self.close(cursor, conn)
     
151.             return count
     
152. 
     
153. 
     
154. if __name__ == '__main__':
     
155.     db = MySqLHelper()
     
156.     # # 查询单条
     
157.     # sql1 = 'select * from userinfo where name=%s'
     
158.     # args = 'python'
     
159.     # ret = db.selectone(sql=sql1, param=args)
     
160.     # print(ret)  # (None, b'python', b'123456', b'0')
     
161.     # 增加单条
     
162.     # sql2 = 'insert into userinfo (name,password) VALUES (%s,%s)'
     
163.     # ret = db.insertone(sql2, ('old2','22222'))
     
164.     # print(ret)
     
165.     # 增加多条
     
166.     # sql3 = 'insert into userinfo (name,password) VALUES (%s,%s)'
     
167.     # li = li = [
     
168.     #     ('分省', '123'),
     
169.     #     ('到达','456')
     
170.     # ]
     
171.     # ret = db.insertmany(sql3,li)
     
172.     # print(ret)
     
173.     # 删除
     
174.     # sql4 = 'delete from  userinfo WHERE name=%s'
     
175.     # args = 'xxxx'
     
176.     # ret = db.delete(sql4, args)
     
177.     # print(ret)
     
178.     # 更新
     
179.     # sql5 = r'update userinfo set password=%s WHERE name LIKE %s'
     
180.     # args = ('993333993', '%old%')
     
181.     # ret = db.update(sql5, args)
     
182.     # print(ret)

复制代码

python3 实现mysql数据库连接池


原理

python编程中可以使用MySQLdb进行数据库的连接及诸如查询/插入/更新等操作，但是每次连接mysql数据库请求时，都是独立的去请求访问，相当浪费资源，
而且访问数量达到一定数量时，对mysql的性能会产生较大的影响。
因此，实际使用中，通常会使用数据库的连接池技术，来访问数据库达到资源复用的目的。
安装数据库连接池模块DBUtils

1. pip3 install 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的连接池对象，并创建了如查询/插入等通用的操作方法。该部分代码实现如下：
还有很多其他参数可以配置：
    dbapi ：数据库接口
    mincached ：启动时开启的空连接数量
    maxcached ：连接池最大可用连接数量
    maxshared ：连接池最大可共享连接数量
    maxconnections ：最大允许连接数量
    blocking ：达到最大数量时是否阻塞
    maxusage ：单个连接最大复用次数
根据自己的需要合理配置上述的资源参数，以满足自己的实际需要。
代码：

1. #!/usr/bin/env python
   
2. # -*- coding:utf-8 -*-
   
3. import pymysql, os, configparser
   
4. from pymysql.cursors import DictCursor
   
5. from DBUtils.PooledDB import PooledDB
   
6. 
   
7. 
   
8. class Config(object):
   
9.     """
   
10.     # Config().get_content("user_information")
    
11. 
    
12.     配置文件里面的参数
    
13.     [notdbMysql]
    
14.     host = 192.168.1.101
    
15.     port = 3306
    
16.     user = root
    
17.     password = python123
    
18.     """
    
19. 
    
20.     def __init__(self, config_filename="myProjectConfig.cnf"):
    
21.         file_path = os.path.join(os.path.dirname(__file__), config_filename)
    
22.         self.cf = configparser.ConfigParser()
    
23.         self.cf.read(file_path)
    
24. 
    
25.     def get_sections(self):
    
26.         return self.cf.sections()
    
27. 
    
28.     def get_options(self, section):
    
29.         return self.cf.options(section)
    
30. 
    
31.     def get_content(self, section):
    
32.         result = {}
    
33.         for option in self.get_options(section):
    
34.             value = self.cf.get(section, option)
    
35.             result[option] = int(value) if value.isdigit() else value
    
36.         return result
    
37. 
    
38. 
    
39. class BasePymysqlPool(object):
    
40.     def __init__(self, host, port, user, password, db_name=None):
    
41.         self.db_host = host
    
42.         self.db_port = int(port)
    
43.         self.user = user
    
44.         self.password = str(password)
    
45.         self.db = db_name
    
46.         self.conn = None
    
47.         self.cursor = None
    
48. 
    
49. 
    
50. class MyPymysqlPool(BasePymysqlPool):
    
51.     """
    
52.     MYSQL数据库对象，负责产生数据库连接 , 此类中的连接采用连接池实现获取连接对象：conn = Mysql.getConn()
    
53.             释放连接对象;conn.close()或del conn
    
54.     """
    
55.     # 连接池对象
    
56.     __pool = None
    
57. 
    
58.     def __init__(self, conf_name=None):
    
59.         self.conf = Config().get_content(conf_name)
    
60.         super(MyPymysqlPool, self).__init__(**self.conf)
    
61.         # 数据库构造函数，从连接池中取出连接，并生成操作游标
    
62.         self._conn = self.__getConn()
    
63.         self._cursor = self._conn.cursor()
    
64. 
    
65.     def __getConn(self):
    
66.         """
    
67.         @summary: 静态方法，从连接池中取出连接
    
68.         @return MySQLdb.connection
    
69.         """
    
70.         if MyPymysqlPool.__pool is None:
    
71.             __pool = PooledDB(creator=pymysql,
    
72.                               mincached=1,
    
73.                               maxcached=20,
    
74.                               host=self.db_host,
    
75.                               port=self.db_port,
    
76.                               user=self.user,
    
77.                               passwd=self.password,
    
78.                               db=self.db,
    
79.                               use_unicode=False,
    
80.                               charset="utf8",
    
81.                               cursorclass=DictCursor)
    
82.         return __pool.connection()
    
83. 
    
84.     def getAll(self, sql, param=None):
    
85.         """
    
86.         @summary: 执行查询，并取出所有结果集
    
87.         @param sql:查询ＳＱＬ，如果有查询条件，请只指定条件列表，并将条件值使用参数[param]传递进来
    
88.         @param param: 可选参数，条件列表值（元组/列表）
    
89.         @return: result list(字典对象)/boolean 查询到的结果集
    
90.         """
    
91.         if param is None:
    
92.             count = self._cursor.execute(sql)
    
93.         else:
    
94.             count = self._cursor.execute(sql, param)
    
95.         if count > 0:
    
96.             result = self._cursor.fetchall()
    
97.         else:
    
98.             result = False
    
99.         return result
    
100. 
     
101.     def getOne(self, sql, param=None):
     
102.         """
     
103.         @summary: 执行查询，并取出第一条
     
104.         @param sql:查询ＳＱＬ，如果有查询条件，请只指定条件列表，并将条件值使用参数[param]传递进来
     
105.         @param param: 可选参数，条件列表值（元组/列表）
     
106.         @return: result list/boolean 查询到的结果集
     
107.         """
     
108.         if param is None:
     
109.             count = self._cursor.execute(sql)
     
110.         else:
     
111.             count = self._cursor.execute(sql, param)
     
112.         if count > 0:
     
113.             result = self._cursor.fetchone()
     
114.         else:
     
115.             result = False
     
116.         return result
     
117. 
     
118.     def getMany(self, sql, num, param=None):
     
119.         """
     
120.         @summary: 执行查询，并取出num条结果
     
121.         @param sql:查询ＳＱＬ，如果有查询条件，请只指定条件列表，并将条件值使用参数[param]传递进来
     
122.         @param num:取得的结果条数
     
123.         @param param: 可选参数，条件列表值（元组/列表）
     
124.         @return: result list/boolean 查询到的结果集
     
125.         """
     
126.         if param is None:
     
127.             count = self._cursor.execute(sql)
     
128.         else:
     
129.             count = self._cursor.execute(sql, param)
     
130.         if count > 0:
     
131.             result = self._cursor.fetchmany(num)
     
132.         else:
     
133.             result = False
     
134.         return result
     
135. 
     
136.     def insertMany(self, sql, values):
     
137.         """
     
138.         @summary: 向数据表插入多条记录
     
139.         @param sql:要插入的ＳＱＬ格式
     
140.         @param values:要插入的记录数据tuple(tuple)/list[list]
     
141.         @return: count 受影响的行数
     
142.         """
     
143.         count = self._cursor.executemany(sql, values)
     
144.         return count
     
145. 
     
146.     def __query(self, sql, param=None):
     
147.         if param is None:
     
148.             count = self._cursor.execute(sql)
     
149.         else:
     
150.             count = self._cursor.execute(sql, param)
     
151.         return count
     
152. 
     
153.     def update(self, sql, param=None):
     
154.         """
     
155.         @summary: 更新数据表记录
     
156.         @param sql: ＳＱＬ格式及条件，使用(%s,%s)
     
157.         @param param: 要更新的  值 tuple/list
     
158.         @return: count 受影响的行数
     
159.         """
     
160.         return self.__query(sql, param)
     
161. 
     
162.     def insert(self, sql, param=None):
     
163.         """
     
164.         @summary: 更新数据表记录
     
165.         @param sql: ＳＱＬ格式及条件，使用(%s,%s)
     
166.         @param param: 要更新的  值 tuple/list
     
167.         @return: count 受影响的行数
     
168.         """
     
169.         return self.__query(sql, param)
     
170. 
     
171.     def delete(self, sql, param=None):
     
172.         """
     
173.         @summary: 删除数据表记录
     
174.         @param sql: ＳＱＬ格式及条件，使用(%s,%s)
     
175.         @param param: 要删除的条件 值 tuple/list
     
176.         @return: count 受影响的行数
     
177.         """
     
178.         return self.__query(sql, param)
     
179. 
     
180.     def begin(self):
     
181.         """
     
182.         @summary: 开启事务
     
183.         """
     
184.         self._conn.autocommit(0)
     
185. 
     
186.     def end(self, option='commit'):
     
187.         """
     
188.         @summary: 结束事务
     
189.         """
     
190.         if option == 'commit':
     
191.             self._conn.commit()
     
192.         else:
     
193.             self._conn.rollback()
     
194. 
     
195.     def dispose(self, isEnd=1):
     
196.         """
     
197.         @summary: 释放连接池资源
     
198.         """
     
199.         if isEnd == 1:
     
200.             self.end('commit')
     
201.         else:
     
202.             self.end('rollback')
     
203.         self._cursor.close()
     
204.         self._conn.close()
     
205. 
     
206. 
     
207. if __name__ == '__main__':
     
208.     mysql = MyPymysqlPool("notdbMysql")
     
209. 
     
210.     sqlAll = "select * from myTest.aa;"
     
211.     result = mysql.getAll(sqlAll)
     
212.     print(result)
     
213. 
     
214.     sqlAll = "select * from myTest.aa;"
     
215.     result = mysql.getMany(sqlAll, 2)
     
216.     print(result)
     
217. 
     
218.     result = mysql.getOne(sqlAll)
     
219.     print(result)
     
220. 
     
221.     # mysql.insert("insert into myTest.aa set a=%s", (1))
     
222. 
     
223.     # 释放资源
     
224.     mysql.dispose()

复制代码

参考博客：https://www.cnblogs.com/renfanzi/p/7656142.html
到此这篇关于python3 实现mysql数据库连接池的示例代码的文章就介绍到这了,更多相关python3 mysql连接池内容请搜索晓枫资讯以前的文章或继续浏览下面的相关文章希望大家以后多多支持晓枫资讯！

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！

您脱风云错

顶顶更健康！！！

巫元

路过，支持一下

如初306

感谢楼主分享。