今天是：2025年6月22日 星期日
记住用户名密码
问题
对于数据库连接，一般不建议使用全局变量，在每次操作完成后即关闭连接。这是因为长时间保持数据库连接会对性能和资源消耗产生负面影响。与此相反，使用数据库连接池来维护和分配数据库连接是更好的做法。
好处
连接池的优点是可以在多个线程或进程之间共享，并且可以有效地管理连接数，而无需手动打开和关闭连接。
常用包
SQLAlchemy 中的 QueuePool 和 DBUtils 中的 PooledDB 都是流行的 Python数据库连接池实现，它们具有相似的功能但具有一些区别。
QueuePool 是 SQLAlchemy 内置的一个连接池实现，它可以管理一个连接队列，确保每个连接在使用后被适当地关闭。该池使用Python 自带的 queue 模块实现，并支持可配置的最大连接数、预处理语句等特性。优点是易于使用，无需其他依赖，并与SQLAlchemy 之间无缝集成。
PooledDB 是 DBUtils 库提供的一个连接池实现，可以与 SQLAlchemy 或其他 Python数据库库一起使用。它支持多种类型的连接池，并使用 threading模块实现线程安全，具有更高的性能和稳定性。该库还提供了一些方便的功能，例如自动回收空闲连接等。
总结以上还是使用 DBUtils 比较好些
代码
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 | import pymysqlfrom dbutils.pooled_db import PooledDBhost = 'localhost'port = 3306user = 'root'password = '123456'database = 'mytest'class MySQLConnectionPool: def __init__(self,): self.pool = PooledDB( creator=pymysql, # 使用链接数据库的模块 mincached=10, # 初始化时，链接池中至少创建的链接，0表示不创建 maxconnections=200, # 连接池允许的最大连接数，0和None表示不限制连接数 blocking=True, # 连接池中如果没有可用连接后，是否阻塞等待。True，等待；False，不等待然后报错 host=host, port=port, user=user, password=password, database=database ) def open(self): self.conn = self.pool.connection() self.cursor = self.conn.cursor(cursor=pymysql.cursors.DictCursor) # 表示读取的数据为字典类型 return self.conn, self.cursor def close(self, cursor, conn): cursor.close() conn.close() def select_one(self, sql, *args): """查询单条数据""" conn, cursor = self.open() cursor.execute(sql, args) result = cursor.fetchone() self.close(conn, cursor) return result def select_all(self, sql, args): """查询多条数据""" conn, cursor = self.open() cursor.execute(sql, args) result = cursor.fetchall() self.close(conn, cursor) return result def insert_one(self, sql, args): """插入单条数据""" self.execute(sql, args, isNeed=True) def insert_all(self, sql, datas): """插入多条批量插入""" conn, cursor = self.open() try: cursor.executemany(sql, datas) conn.commit() return {'result': True, 'id': int(cursor.lastrowid)} except Exception as err: conn.rollback() return {'result': False, 'err': err} def update_one(self, sql, args): """更新数据""" self.execute(sql, args, isNeed=True) def delete_one(self, sql, *args): """删除数据""" self.execute(sql, args, isNeed=True) def execute(self, sql, args, isNeed=False): """ 执行 :param isNeed 是否需要回滚 """ conn, cursor = self.open() if isNeed: try: cursor.execute(sql, args) conn.commit() except: conn.rollback() else: cursor.execute(sql, args) conn.commit() self.close(conn, cursor)"""CREATE TABLE `names` ( `id` int(10) NOT NULL AUTO_INCREMENT COMMENT '主键', `name` VARCHAR(30) DEFAULT NULL COMMENT '姓名', `sex` VARCHAR(20) DEFAULT NULL COMMENT '性别', `age` int(5) DEFAULT NULL COMMENT '年龄', PRIMARY KEY (`id`) USING BTREE) ENGINE=InnoDB DEFAULT CHARSET=utf8 ROW_FORMAT=DYNAMIC COMMENT='数据导入mysql';"""mysql = MySQLConnectionPool()sql_insert_one = "insert into `names` (`name`, sex, age) values (%s,%s,%s)"mysql.insert_one(sql_insert_one, ('唐三', '男', 25))datas = [ ('戴沐白', '男', 26), ('奥斯卡', '男', 26), ('唐三', '男', 25), ('小舞', '女', 100000), ('马红俊', '男', 23), ('宁荣荣', '女', 22), ('朱竹清', '女', 21),]sql_insert_all = "insert into `names` (`name`, sex, age) values (%s,%s,%s)"mysql.insert_all(sql_insert_all, datas)sql_update_one = "update `names` set age=%s where `name`=%s"mysql.update_one(sql_update_one, (28, '唐三'))sql_delete_one = 'delete from `names` where `name`=%s 'mysql.delete_one(sql_delete_one, ('唐三',))sql_select_one = 'select * from `names` where `name`=%s'results = mysql.select_one(sql_select_one, ('唐三',))print(results)sql_select_all = 'select * from `names` where `name`=%s'results = mysql.select_all(sql_select_all, ('唐三',))print(results) |
目前有 0 条留言 其中：访客:0 条, 博主:0 条