本文目录导读:

对于Python数据库连接池,DBUtils 是一个经典选择,但需要根据你的需求和Python版本综合考虑。
DBUtils 的特点
优点:
- 成熟稳定,历史悠久
- 支持多种数据库后端(MySQL、PostgreSQL、SQLite等)
- 提供两种模式:
PersistentDB(线程专用)和PooledDB(线程共享) - 自动管理连接的生命周期
缺点:
- 项目维护频率较低(最后更新较久)
- 对异步编程支持较弱
- 在某些Python新版本中可能存在兼容性问题
实际使用示例
from DBUtils.PooledDB import PooledDB
import pymysql
# 创建连接池
pool = PooledDB(
creator=pymysql, # 使用PyMySQL
maxconnections=6, # 最大连接数
mincached=2, # 初始化时创建的连接数
maxcached=5, # 最大空闲连接数
blocking=True, # 无连接时是否阻塞等待
maxusage=None, # 单个连接最大复用次数
setsession=[], # 会话设置
host='localhost',
port=3306,
user='root',
password='password',
database='test',
charset='utf8mb4'
)
# 使用连接
def get_user(id):
conn = pool.connection()
try:
with conn.cursor() as cursor:
cursor.execute("SELECT * FROM users WHERE id=%s", (id,))
return cursor.fetchone()
finally:
conn.close() # 实际是归还到连接池
更现代的替代方案
SQLAlchemy(推荐)
from sqlalchemy import create_engine
from sqlalchemy.pool import QueuePool
# 内置连接池
engine = create_engine(
'mysql+pymysql://user:password@localhost/test',
poolclass=QueuePool,
pool_size=5,
max_overflow=10,
pool_pre_ping=True # 连接检测
)
psycopg2(PostgreSQL专用)
from psycopg2 import pool
connection_pool = pool.SimpleConnectionPool(
1, 20,
user='user',
password='password',
host='localhost',
database='test'
)
aiomysql/asyncpg(异步方案)
import asyncio
import aiomysql
async def main():
pool = await aiomysql.create_pool(
host='localhost',
port=3306,
user='root',
password='password',
db='test',
minsize=1,
maxsize=10
)
async with pool.acquire() as conn:
async with conn.cursor() as cursor:
await cursor.execute("SELECT 1")
result = await cursor.fetchone()
选择建议
| 场景 | 推荐方案 |
|---|---|
| 传统同步Web应用 | SQLAlchemy(内置连接池) |
| 已有DBUtils代码 | 继续使用,但考虑迁移 |
| 轻量级脚本 | DBUtils 或 直接使用SQLAlchemy |
| 异步应用 | asyncpg (PG) / aiomysql (MySQL) |
| 需要ORM | SQLAlchemy + Autoflush |
注意事项
- 不要混用:避免在同一项目中同时使用多种连接池方案
- 连接测试:设置
pool_pre_ping=True防止使用失效连接 - 资源管理:确保使用
try/finally或with语句归还连接 - 监控:生产环境建议监控连接池使用情况
DBUtils仍然可用,但如果是新项目,更推荐使用SQLAlchemy内置的连接池,它更现代、维护更活跃,且提供了完整的ORM功能。