Python3 数据库连接池：PyMySQL 驱动配置与实战应用

一、连接池核心价值

1. 性能优化：复用连接减少建立/断开开销
2. 资源管控：防止连接数爆炸($$ \text{max_connections} \leq \text{pool_size} $$)
3. 异常恢复：自动处理失效连接

二、环境配置

pip install PyMySQL DBUtils  # 核心依赖

三、连接池配置详解

from dbutils.pooled_db import PooledDB
import pymysql

# 连接池配置
POOL = PooledDB(
    creator=pymysql,  # 驱动类型
    maxconnections=10,  # 最大连接数
    mincached=2,       # 初始空闲连接
    host='localhost',
    user='root',
    password='your_pwd',
    database='test_db',
    charset='utf8mb4',
    autocommit=False   # 建议手动提交事务
)

四、实战应用场景

1. 基础查询操作

def query_data(sql):
    conn = POOL.connection()
    try:
        with conn.cursor() as cursor:
            cursor.execute(sql)
            return cursor.fetchall()
    finally:
        conn.close()  # 实际归还连接至池

1. 事务处理模式

def update_transaction():
    conn = POOL.connection()
    try:
        with conn.cursor() as cursor:
            cursor.execute("UPDATE users SET balance=balance-100 WHERE id=1")
            cursor.execute("UPDATE shops SET revenue=revenue+100 WHERE id=5")
        conn.commit()  # 显式提交
    except Exception as e:
        conn.rollback()  # 事务回滚
        raise e
    finally:
        conn.close()

1. 连接健康检查（关键！）

def check_connection(conn):
    try:
        with conn.cursor() as cursor:
            cursor.execute("SELECT 1")
            return cursor.fetchone()[0] == 1
    except:
        return False  # 标记失效连接

五、性能优化建议

1. 连接复用公式： $$ \text{复用率} = \frac{\text{请求数}}{\text{新建连接数}} \geq 5 $$
2. 超时配置：

   POOL = PooledDB(
       ...,
       ping=1,  # 每次使用前检查连接(0=从不, 1=请求时, 2=创建时)
       wait_timeout=300  # 连接最长空闲时间(秒)
   )
   

3. 监控指标：
   • 活跃连接数：POOL._connections
   • 等待队列：POOL._waiting

六、典型错误处理

try:
    conn = POOL.connection(timeout=3)  # 获取连接超时
except Queue.Empty:
    # 连接池耗尽处理方案
    logger.error("连接池过载！当前等待队列：%s", POOL._waiting)
    return {"code": 503, "msg": "服务繁忙"}

最佳实践：在Web服务中，建议在请求入口获取连接，出口关闭连接，确保每个请求独立使用连接。连接池配置参数应根据实际压测结果($$ QPS \times \text{平均响应时间} $$)动态调整。