SQLAlchemy连接池失效排查：如何优化配置实现有效连接池？

数据老陈

2026-6-18

SQLAlchemy连接池配置未生效导致Stale Connections问题分析与解决

问题核心

配置pool_size=10、max_overflow=20、pool_timeout=10、pool_recycle=250（小于数据库wait_timeout）以及pool_pre_ping=True等连接池参数后，仍出现 stale connections 问题，根源在于代码中存在全局会话滥用和重复引擎实例两个关键问题。

代码问题分析

全局单例Session误用
在Models/models.py中创建了全局的session = Session()实例，该会话会被所有Flask请求共享。由于会话长时间持有同一个数据库连接且不会主动放回连接池，当连接被数据库端超时回收后，再次使用该会话就会触发stale connections错误。
重复创建引擎实例
代码在两处（独立的创建引擎代码和models.py）分别调用create_engine生成引擎，导致存在两个独立的连接池，不仅浪费资源，也会让配置的连接池参数无法统一生效。

修复方案

1. 移除全局Session，改用请求级会话

修改Models/models.py，仅保留Session工厂类，删除全局会话实例：

# Models/models.py 修改后
import os
from datetime import datetime
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import SmallInteger, Column, Integer, String, Float, DateTime
from config import engine  # 后续统一从配置文件导入引擎

Base = declarative_base()

class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    username = Column(String(1000), nullable=False)
    password = Column(String(1000), nullable=False)
    client = Column(String(1000), nullable=False)
    jwt_token = Column(String(2000))
    is_enable = Column(SmallInteger)

Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)
# 移除全局session = Session()这一行

在main.py中通过Flask请求钩子，为每个请求创建独立会话，请求结束后关闭：

# main.py 修改后
# ... 其他导入 ...
from flask import g  # 新增导入g对象
from Models.models import Session

# ... 初始化Flask app等代码 ...

@app.before_request
def create_session():
    # 为每个请求创建会话，存储在g对象中
    g.session = Session()

@app.teardown_request
def close_session(exception=None):
    # 请求结束后关闭会话，将连接放回连接池
    session = getattr(g, 'session', None)
    if session is not None:
        if exception:
            session.rollback()
        session.close()

@app.route(f'{BASE_URL}/list_users', methods=['GET'])
def list_user():
    try:
        # 使用g对象中的请求级会话
        user_list = service_layer.get_all_users(g.session)
        return_data = jsonify({'users': user_list})
        return return_data
    except Exception as e:
        g.session.rollback()
        logger.info(f"Some error while Listing Users - Try Again !! - {e}")
        return make_response("Some error while Listing Users - Try Again - Contact Admin!!", 401)

2. 统一引擎创建，避免重复实例

创建单独的配置文件config.py，集中管理引擎：

# config.py
from sqlalchemy import create_engine

DB_URL = "db_string"
engine = create_engine(DB_URL,
                       pool_size=10,
                       max_overflow=20,
                       pool_timeout=10,
                       pool_recycle=250,
                       pool_pre_ping=True
                       )

后续所有需要引擎的地方（如models.py）都从config.py导入，确保只有一个引擎实例和对应的连接池。

3. 验证连接池配置有效性

开启SQLAlchemy的连接池日志，确认连接回收、ping机制是否正常运作：

# 在main.py或config.py中添加日志配置
import logging
logging.basicConfig(level=logging.INFO)
logging.getLogger('sqlalchemy.engine').setLevel(logging.INFO)
logging.getLogger('sqlalchemy.pool').setLevel(logging.INFO)

通过日志可以观察到连接池的连接获取、释放、回收以及pool_pre_ping的执行情况，验证配置是否生效。

4. 确认数据库wait_timeout设置

确保pool_recycle的值确实小于数据库的wait_timeout（例如数据库wait_timeout设为300秒，pool_recycle设为250秒），避免数据库主动回收连接后，连接池仍认为连接有效。

内容的提问来源于stack exchange，提问作者Ginni Garg