如何在 Python 中检测 MySQL 数据库查询超时-Python教程-PHP中文网

如何在 Python 中检测 MySQL 数据库查询超时

在使用 MySQL Connector/Python 时，可通过 max_execution_time 设置查询执行时限，并捕获 OperationalError 异常中包含“Query execution was interrupted”的特定错误信息，从而准确识别并处理查询超时事件。

在使用 mysql connector/python 时，可通过 `max_execution_time` 设置查询执行时限，并捕获 `operationalerror` 异常中包含“query execution was interrupted”的特定错误信息，从而准确识别并处理查询超时事件。

在 Python 应用中对数据库查询设置超时，是保障服务响应性与用户体验的关键实践。MySQL 5.7.8+ 支持通过会话变量 max_execution_time（单位：毫秒）限制单条 SELECT 查询的最长执行时间。但需注意：该机制不会静默失败，而是主动中断查询并抛出可捕获的异常——关键在于正确识别该异常类型及错误消息。

以下是一个健壮、可复用的超时检测示例：

import mysql.connector
from mysql.connector import Error

def execute_with_timeout(cursor, query, timeout_ms=1000):
    try:
        # 重置会话并设置查询超时（仅对后续 SELECT 生效）
        cursor._connection.reset_session(
            session_variables={'max_execution_time': timeout_ms}
        )
        cursor.execute(query)

        # 尝试获取结果（重要！超时实际在 fetch 阶段触发）
        results = cursor.fetchall()
        return {"success": True, "data": results}

    except mysql.connector.errors.OperationalError as e:
        error_msg = str(e).strip()
        # MySQL 官方错误信息（区分大小写敏感匹配更可靠）
        if "Query execution was interrupted" in error_msg or "Query timeout" in error_msg:
            return {"success": False, "error": "query_timeout", "message": "数据库查询已超时"}
        else:
            return {"success": False, "error": "operational_error", "message": error_msg}

    except mysql.connector.Error as e:
        return {"success": False, "error": "mysql_error", "message": str(e)}

    finally:
        # 注意：cursor 不应在此处 close，避免影响复用；由调用方统一管理生命周期
        pass

# 使用示例
try:
    conn = mysql.connector.connect(
        host='localhost',
        user='your_user',
        password='your_pass',
        database='testdb'
    )
    cursor = conn.cursor(buffered=False)

    # 测试超时（SLEEP(2) > max_execution_time=1000ms）
    result = execute_with_timeout(cursor, "SELECT SLEEP(2)", timeout_ms=1000)

    if result["success"]:
        print("✅ 查询成功:", len(result["data"]), "行数据")
    else:
        if result["error"] == "query_timeout":
            print("⏰ 检测到查询超时 —— 请优化SQL或延长超时阈值")
        else:
            print("❌ 数据库错误:", result["message"])

finally:
    if 'cursor' in locals():
        cursor.close()
    if 'conn' in locals() and conn.is_connected():
        conn.close()

? 关键注意事项：

MySQL8 CRM Schema Design Expert

面向 CRM 系统的 MySQL 8 数据库模式设计。当用户需要为客户关系管理（CRM）系统设计、评审、优化或生成数据库模式时，应使用此项技能。

下载

max_execution_time 仅对 SELECT 语句生效，INSERT/UPDATE/DELETE 不受其约束（MySQL 行为）；如需控制写操作超时，应使用连接层 connection_timeout 或 socket_timeout 参数。
超时异常通常在 cursor.execute() 或 cursor.fetchall() 时抛出，务必在获取结果时也包裹异常处理，避免漏判。
错误消息字符串依赖 MySQL 版本语言配置（如中文版可能显示“查询执行被中断”），生产环境建议结合 e.errno == 3024（MySQL 错误码：ER_QUERY_TIMEOUT）进行双重判断，提升健壮性：
```
except mysql.connector.errors.OperationalError as e:
    if e.errno == 3024:  # 官方超时错误码
        handle_timeout()
```