上一篇
Flask如何使用DBUtils连接数据库，常见问题有哪些？
环境准备
使用 DBUtils 需要先安装 Flask 和数据库驱动（如 pymysql 或 psycopg2）以及 DBUtils 本身。
- Python 3.6+
- Flask 2.x
- DBUtils 3.0+
- 数据库驱动（以 MySQL 为例：pymysql）
安装依赖
pip install flask dbutils pymysql
DBUtils 连接池简介
DBUtils 提供两种主要连接池：
| 连接池类型 | 特点 | 适用场景 |
|---|---|---|
| PersistentDB | 每个线程维护一个专用连接，线程安全，连接数量固定 | 单线程或多线程线程数较少且稳定 |
| PooledDB | 共享连接池，支持动态调整连接数，线程安全 | 多线程频繁请求，连接数波动大 |
配置数据库连接池
以 PooledDB 为例，创建连接池模块 db_pool.py：
from dbutils.pooled_db import PooledDB import pymysql POOL = PooledDB( creator=pymysql, # 使用 pymysql 连接 maxconnections=10, # 连接池最大连接数 mincached=2, # 初始化时最少空闲连接数 maxcached=5, # 最大空闲连接数 blocking=True, # 无可用连接时是否阻塞等待 maxusage=None, # 单个连接最大复用次数，None 表示不限制 setsession=[], # 执行前执行的 SQL 列表 ping=0, # 检查连接活性的间隔（秒），0 表示不检查 host='localhost', port=3306, user='root', password='password', database='test', charset='utf8' )
在 Flask 中使用连接池
在 Flask 应用中导入连接池，通过 get_connection() 获取连接，使用后释放（归还给池）。
from flask import Flask, jsonify from db_pool import POOL app = Flask(__name__) @app.route('/user/<int:user_id>') def get_user(user_id): conn = POOL.connection() cursor = conn.cursor() try: cursor.execute("SELECT id, name FROM users WHERE id=%s", (user_id,)) row = cursor.fetchone() if row: return jsonify({"id": row[0], "name": row[1]}) return jsonify({"error": "not found"}), 404 finally: cursor.close() conn.close() # 释放连接，归还到池中 if __name__ == '__main__': app.run(debug=True)
完整示例：带事务的更新操作
@app.route('/update', methods=['POST']) def update_user(): data = request.get_json() user_id = data['id'] new_name = data['name'] conn = POOL.connection() cursor = conn.cursor() try: conn.begin() # 开始事务 cursor.execute("UPDATE users SET name=%s WHERE id=%s", (new_name, user_id)) conn.commit() return jsonify({"status": "ok"}) except Exception: conn.rollback() return jsonify({"status": "error"}), 500 finally: cursor.close() conn.close()
使用 PersistentDB 的注意事项
- PersistentDB 适用于线程数固定的场景，每个线程独立连接,不需要显式关闭。
- 在 Flask 默认线程池下，线程会复用，但 PersistentDB 的线程‑连接绑定可能导致连接泄漏，因此推荐使用 PooledDB。
相关问题与解答
问题 1：DBUtils 连接池中的连接是否需要手动关闭？
- 答：需要，使用 conn.close() 并不会真正关闭连接，而是将连接归还到连接池中，供后续复用，如果不调用 close()，连接会被一直占用,可能导致连接池耗尽。
问题 2：在多线程 Flask 应用中，PooledDB 的 blocking=True 会导致请求阻塞吗？
- 答：会，当连接池中的连接全部被占用时，blocking=True 会使当前线程等待直到有可用连接，如果等待时间过长，可能影响响应速度，建议设置合理的 maxconnections 并根据业务调整 maxcached 和 mincached，必要时配合超时设置（如 timeout 参数）。
