跳到主要内容

连接池

问题​

为什么需要数据库连接池？Python 中有哪些连接池方案？

答案​

数据库连接的创建（TCP 三次握手 + 认证）开销大，连接池复用已建立的连接，避免频繁创建销毁。

SQLAlchemy 连接池​

SQLAlchemy 内置 QueuePool：

from sqlalchemy import create_engine

engine = create_engine(
    "postgresql://user:pass@localhost/db",
    pool_size=10,          # 池中保持的连接数
    max_overflow=20,       # 超出 pool_size 后允许的临时连接数
    pool_timeout=30,       # 获取连接的最大等待秒数
    pool_recycle=1800,     # 连接回收时间（秒），避免数据库侧超时
    pool_pre_ping=True,    # 使用前 ping 检测连接是否存活
)

异步连接池​

from sqlalchemy.ext.asyncio import create_async_engine

# asyncpg 自带连接池
engine = create_async_engine(
    "postgresql+asyncpg://user:pass@localhost/db",
    pool_size=20,
    max_overflow=10,
)
# 或直接使用 asyncpg 连接池
import asyncpg

pool = await asyncpg.create_pool(
    "postgresql://user:pass@localhost/db",
    min_size=5,
    max_size=20,
    max_inactive_connection_lifetime=300,
)

async with pool.acquire() as conn:
    row = await conn.fetchrow("SELECT * FROM users WHERE id = $1", user_id)

连接池状态监控​

# SQLAlchemy 连接池事件
from sqlalchemy import event

@event.listens_for(engine, "checkout")
def on_checkout(dbapi_conn, connection_record, connection_proxy):
    """连接从池中取出时触发"""
    pass

@event.listens_for(engine, "checkin")
def on_checkin(dbapi_conn, connection_record):
    """连接归还到池中时触发"""
    pass

# 查看连接池状态
pool = engine.pool
print(f"Size: {pool.size()}, Checked out: {pool.checkedout()}")

常见面试问题​

Q1: pool_size 设多大合适？​

答案：

经验公式：pool_size = CPU 核心数 × 2 + 磁盘数（PostgreSQL 推荐）。Web 应用中通常设为 Worker 数 × 2~5。关键是监控连接等待时间，如果频繁超时说明池太小。

Q2: pool_recycle 有什么用？​

答案：

MySQL 默认 8 小时后关闭空闲连接（wait_timeout），如果连接池中的连接超过这个时间未使用，再取出时会报 MySQL server has gone away。pool_recycle=3600 会在连接存活 1 小时后主动回收。

Q3: pool_pre_ping 的原理？​

答案：

每次从池中取连接时，先发一个 SELECT 1 验证连接是否存活。如果失联则丢弃该连接并创建新连接。有轻微性能开销，但能彻底避免断连问题。

相关链接​