Flask应用中如何缓存MySQL连接以优化请求性能？

嘿，这个问题我太懂了——频繁创建数据库连接真的是性能杀手，尤其是API请求量上来的时候。还好MySQL的Python驱动和Flask生态都有现成的解决方案，不用自己造轮子。下面给你两种最常用的实现方式：

方法一：用mysql-connector-python自带的连接池 #

既然你已经在用mysql.connector，直接用它内置的连接池是最省心的，不需要额外装库。

步骤1：初始化连接池 #

在Flask应用启动的时候创建连接池，而不是每次请求创建连接：

from flask import Flask
import mysql.connector
from mysql.connector import pooling

app = Flask(__name__)

# 创建连接池，参数可以根据你的需求调整
db_pool = mysql.connector.pooling.MySQLConnectionPool(
    pool_name="flask_mysql_pool",
    pool_size=5,  # 建议根据并发量设置，比如5-10，别太大
    pool_reset_session=True,  # 归还连接时重置会话，避免遗留状态
    user='xx',
    password='xx',
    host='xx',
    database='xx'
)

步骤2：在接口中使用连接池 #

每次请求从池子里拿连接，用完记得归还（调用close()其实是放回池子，不是真的关闭连接）：

@app.route('/api/get-data')
def get_data():
    # 从连接池获取连接
    cnx = db_pool.get_connection()
    cursor = cnx.cursor(dictionary=True)  # 返回字典格式结果，前端更友好
    
    try:
        cursor.execute("SELECT * FROM your_table LIMIT 10")
        result = cursor.fetchall()
        return {"code": 200, "data": result}
    except Exception as e:
        return {"code": 500, "msg": str(e)}
    finally:
        # 不管成功失败，都要关闭游标和归还连接
        cursor.close()
        cnx.close()

小提示：用钩子兜底 #

怕有时候忘记归还连接？可以加个请求结束的钩子，确保连接不会泄漏：

@app.teardown_request
def teardown_db(exception):
    # 这里可以针对特殊情况做处理，不过上面的finally已经足够靠谱
    pass

方法二：用Flask-SQLAlchemy（更适合长期维护的项目） #

如果你的项目打算长期迭代，推荐用Flask-SQLAlchemy——它不仅自带连接池，还提供ORM功能，能大幅减少原生SQL的重复代码，而且连接管理完全自动化。

步骤1：安装依赖 #

pip install flask-sqlalchemy

步骤2：配置并初始化 #

from flask import Flask
from flask_sqlalchemy import SQLAlchemy

app = Flask(__name__)

# 配置数据库连接URI，格式是：mysql+mysqlconnector://用户名:密码@主机/数据库名
app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql+mysqlconnector://xx:xx@xx/xx'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False  # 关闭不必要的修改跟踪，节省资源
app.config['SQLALCHEMY_POOL_SIZE'] = 5  # 连接池大小
app.config['SQLALCHEMY_POOL_RECYCLE'] = 3600  # 1小时自动回收连接，防止MySQL断开闲置连接

db = SQLAlchemy(app)

步骤3：使用连接池（ORM或原生SQL都支持） #

方式一：用ORM查询（推荐）

先定义模型对应数据库表：

class YourTable(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    title = db.Column(db.String(100))
    content = db.Column(db.Text)

然后在接口中查询：

@app.route('/api/get-data')
def get_data():
    # ORM方式查询，自动处理连接
    items = YourTable.query.limit(10).all()
    # 转成字典返回给客户端
    result = [{
        "id": item.id,
        "title": item.title,
        "content": item.content
    } for item in items]
    return {"code": 200, "data": result}

方式二：用原生SQL

如果习惯写原生SQL，也可以直接用：

@app.route('/api/get-data')
def get_data():
    result = db.session.execute("SELECT * FROM your_table LIMIT 10").fetchall()
    # 把结果转成字典
    data = [dict(row._mapping) for row in result]
    return {"code": 200, "data": data}

关键注意事项 #

• 连接池大小别乱设：太小会导致请求等待连接，太大则会占用过多服务器资源，建议根据你的并发量（比如QPS）调整，一般5-20之间比较合适。
• 处理闲置连接：MySQL服务器会自动断开长时间闲置的连接，所以要设置连接回收时间（比如SQLAlchemy的SQLALCHEMY_POOL_RECYCLE，或者mysql-connector的pool_reset_session），避免拿到失效连接报错。
• 别用全局单连接：Flask是多线程的，单个连接在多线程下会出现并发问题，连接池是线程安全的，所以一定要用池来管理连接。

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