Flask+SQLAlchemy无法限制MySQL连接数，求test2场景配置方案

问题根源 #

你在test2场景下的核心问题是：每次请求都通过before_request新建create_engine实例。SQLAlchemy中每个Engine对应一个独立的连接池（默认QueuePool），反复创建Engine等于每个请求都生成一个全新的连接池，配置的pool_size等参数只对单个池生效，但总连接数会随着请求数累加，完全起不到全局限制的作用。

解决方案 #

要限制全局连接数，必须保证所有请求共用同一个Engine实例，基于这个共用的Engine来创建Scoped Session。修改步骤如下：

• 全局初始化Engine和连接池：将Engine的创建移到全局作用域，一次性配置好连接池参数，所有请求复用这个Engine。
• 复用Scoped Session：Scoped Session本身就是为多线程/请求场景设计的，会自动为每个请求分配独立的Session实例，不需要在before_request里重新创建。
• 修正请求销毁逻辑：确保teardown_appcontext中正确移除当前请求的Session，让连接归还到连接池。

修改后的代码示例 #

from flask import Flask
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.orm import sessionmaker, scoped_session
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.pool import QueuePool

app = Flask(__name__)

# 全局初始化Engine，配置连接池参数
engine = create_engine(
    "mysql+pymysql://root:password@localhost:3306/mytestdb",
    pool_size=10,          # 核心池维护的常驻连接数
    max_overflow=20,       # 核心池满后允许临时创建的额外连接数
    pool_recycle=3600      # 自动回收闲置连接，避免MySQL主动断开
)
# 全局创建Scoped Session，自动绑定请求上下文
db_session = scoped_session(sessionmaker(bind=engine))

Base = declarative_base()

class User(Base):
    __tablename__= 'user'
    id = Column(Integer, primary_key=True)
    name = Column(String)
    age = Column(Integer)
    city = Column(String)

@app.route('/', methods=['GET'])
def home():
    # 直接使用全局scoped session，自动关联当前请求
    datas = db_session.query(User).all()
    print([data.name for data in datas])
    return "Success" + str([(data.name, data.age, data.city )for data in datas])

@app.teardown_appcontext
def shutdown(exception=None):
    # 移除当前请求的session实例，连接归还到连接池
    db_session.remove()

if __name__== "__main__":
    app.run(debug=True)

关键说明 #

• 原来的pool_size无效的原因：每个请求新建Engine，每个Engine的池都是独立的，总连接数是所有请求池的连接数之和，根本没有全局复用。
• Scoped Session的作用：它会为每个Flask请求上下文维护一个独立的Session实例，请求结束后调用remove()会将Session占用的连接归还到共用的连接池，实现连接复用。
• pool_recycle参数：建议加上，避免MySQL因闲置时间过长主动断开连接（可根据MySQL的wait_timeout配置调整数值）。

内容的提问来源于stack exchange，提问作者pradeepmaurya-neo