Flask API在cursor.execute时静默崩溃的问题排查与解决

我开发了一个给React应用提供多数据库数据的简易Flask API，原本每次请求都会创建并关闭数据库连接，最近改成启动时创建持久化连接并存入字典后出现异常：

• 偶尔在get_sites和get_assets函数的cursor.execute调用时抛出Lost connection to MySQL server during query错误
• 多数情况是静默崩溃，浏览器控制台显示ERR_CONNECTION_RESET

排查后确认：React组件会同时异步调用两个相关API路由，导致同一个数据库连接被并发查询复用，直接引发Flask崩溃。目前用串行调用能临时解决，但这不是最优方案。

1. 使用数据库连接池（推荐） #

用连接池管理持久化连接，每个请求从池里获取独立连接，用完归还，彻底避免并发冲突。以mysql-connector-python为例：

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

app = Flask(__name__)

# 配置连接池
db_config = {
    'user': 'your_username',
    'password': 'your_password',
    'host': 'your_host',
    'database': 'your_db'
}
connection_pool = pooling.MySQLConnectionPool(
    pool_name="app_pool",
    pool_size=5,  # 根据并发量调整
    **db_config
)

# 路由中使用连接池
@app.route('/get_sites')
def get_sites():
    conn = connection_pool.get_connection()
    try:
        cursor = conn.cursor()
        cursor.execute("SELECT * FROM sites")
        result = cursor.fetchall()
        return jsonify(result)
    finally:
        cursor.close()
        conn.close()  # 归还到连接池，并非真正关闭连接

@app.route('/get_assets')
def get_assets():
    conn = connection_pool.get_connection()
    try:
        cursor = conn.cursor()
        cursor.execute("SELECT * FROM assets")
        result = cursor.fetchall()
        return jsonify(result)
    finally:
        cursor.close()
        conn.close()

2. 为每个请求分配专属连接 #

利用Flask的请求上下文钩子，确保每个请求拥有独立的数据库连接：

from flask import Flask, g, jsonify
import mysql.connector

app = Flask(__name__)
db_config = {
    'user': 'your_username',
    'password': 'your_password',
    'host': 'your_host',
    'database': 'your_db'
}

# 请求开始前创建连接，存入g对象（每个请求独立）
@app.before_request
def before_request():
    g.db_conn = mysql.connector.connect(**db_config)

# 请求结束后关闭连接
@app.teardown_request
def teardown_request(exception):
    if hasattr(g, 'db_conn'):
        g.db_conn.close()

# 路由中使用g对象的专属连接
@app.route('/get_sites')
def get_sites():
    cursor = g.db_conn.cursor()
    cursor.execute("SELECT * FROM sites")
    result = cursor.fetchall()
    cursor.close()
    return jsonify(result)

@app.route('/get_assets')
def get_assets():
    cursor = g.db_conn.cursor()
    cursor.execute("SELECT * FROM assets")
    result = cursor.fetchall()
    cursor.close()
    return jsonify(result)

3. 临时过渡：给共享连接加锁 #

如果暂时无法重构连接管理逻辑，可以给每个共享连接配一把锁，同一时间只允许一个查询使用：

from flask import Flask, jsonify
import mysql.connector
import threading

app = Flask(__name__)

# 共享连接与对应锁
shared_connections = {
    'main_db': mysql.connector.connect(**db_config)
}
conn_locks = {
    'main_db': threading.Lock()
}

@app.route('/get_sites')
def get_sites():
    with conn_locks['main_db']:
        cursor = shared_connections['main_db'].cursor()
        cursor.execute("SELECT * FROM sites")
        result = cursor.fetchall()
        cursor.close()
        return jsonify(result)

@app.route('/get_assets')
def get_assets():
    with conn_locks['main_db']:
        cursor = shared_connections['main_db'].cursor()
        cursor.execute("SELECT * FROM assets")
        result = cursor.fetchall()
        cursor.close()
        return jsonify(result)

注：此方案会导致请求排队，影响接口性能，仅作为临时过渡使用。

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