当前位置：首页 > article >正文

【Python】Flask报错：TimeoutError: QueuePool limit of size 10...以及日常bug处理

article 2025/1/8 12:51:05

1.问题背景

最近在使用Flask框架的flask_sqlalchemy时，突然遇到这个错误：

QueuePool limit of size 5 overflow 10 reached, connection timed out, timeout 30.00 (Background on this error at: https://sqlalche.me/e/20/3o7r)

问题出来，刚开始无从下手，以为框架出问题了，经过对官方的研究和网友们的提示，最终整理一下，为以后遇到此类问题做个积累。

2.问题分析：

产生该问题的原因是因为 SQLAlchemy 连接池的配置问题、并发太高等都可能引起该错误，

还有一个典型的错误原因是，在客户端发送一个请求后， flask后台执行一个数据库操作，随后进行某种逻辑操作，如果该逻辑出现错误，会为客户端返回500错误码，此时该连接未释放，因此会导致数据库连接池继续占用。

2.1 方式一：增加数据库连接池

可以增加数据库连接池的数量进行更新，更改的方式如下：

from flask import Flask
app = Flask(__name__)

# 解决跨域
CORS(app)

app.config['SQLALCHEMY_DATABASE_URI'] = f"mysql+pymysql://root:{encoded_password}@127.0.0.1:3306/test"
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
app.config['SQLALCHEMY_POOL_SIZE'] = 100
# 自动提交映射数据
app.config['SQLALCHEMY_COMMIT_ON_TEARDOWN'] = True

其中这个是核心，这个就是解决这个问题的方式之一：（本例中将数据库连接池改为100，大家根据自己的需求进行相应的更新）

app.config['SQLALCHEMY_POOL_SIZE'] = 100

提供一下官方的地址，可以进行查看

Configuration — Flask-SQLAlchemy Documentation (3.1.x)

Flask的SQLAlchemy — Flask Documentation (3.0.x)

SQLAlchemy - The Database Toolkit for Python

如果采用 db.create_engine 这种方式，没有设置pool_size的大小，默认为5。加上pool_size=100后，此问题不再出现。可以这么设置：

engin = db.create_engine("mysql://root:password@localhost:3306/dbname?charset=utf8", pool_size=100, max_overflow=20)

2.2 方案二：增加try_except

    try:
        r = requests.get('http://github.com', timeout=0.001)
    except:
        r = None

https://blog.csdn.net/weixin_42474540/article/details/105087429

3.问题解决后，再提几个细节上的内容

import urllib.parse

password = "abc@123"
encoded_password = urllib.parse.quote(password)

然后这么去使用：

app.config['SQLALCHEMY_DATABASE_URI'] = f"mysql+pymysql://root:{encoded_password}@127.0.0.1:3306/test"

这样就能避免当你密码中带特殊符号的特殊情况了

https://blog.csdn.net/qq_41020281/article/details/82433096

https://blog.csdn.net/byna11sina11/article/details/120343803

4.线程池样例编写

# 创建线程池，设置最大线程数（根据机器性能、IO密集型还是CPU密集型进行合理设置线程数的大小）
vector_executor = ThreadPoolExecutor(max_workers=5)
# 线程池方法调用
futures = [vector_executor.submit(vector_task, [id]) for id in ids]


# 异步执行向量化,定义线程池和任务函数
def vector_task(id_tuple):
        pass

查看全文

http://www.kler.cn/a/471558.html