前言

在基于 Python 的 Web 应用开发里，数据库连接是极为关键的一环。不过，像网络波动、数据库服务器维护这类因素，都可能造成数据库长时间断连，进而影响应用的正常运作。本文将详细介绍怎样运用 retry_on_failure 装饰器来解决数据库长时间断连的难题

一问题背景

在实际开发场景中，应用和数据库之间的连接可能会由于各种缘由中断（长时间系统无人访问，再次访问，数据库连接超时）。当应用尝试执行数据库操作时，要是连接已经断开，就会抛出异常，致使请求失败。从用户角度来看，这或许表现为页面加载失败或者操作无响应，极大地影响用户体验。所以，我们需要一种机制来应对数据库断连的状况，尽可能恢复连接并重新执行操作。

二程序层面解决方案：retry_on_failure 装饰器

装饰器是 Python 中强大的语法糖，它能够在不修改原有函数代码的基础上，为函数增添额外功能。我们可以借助装饰器实现数据库操作的重试机制，当数据库连接断开时，自动重试一定次数，提升操作成功的几率。

实现代码

下面是 retry_on_failure 装饰器的实现代码，同时包含了 Web 框架和数据库连接库的基本配置：

from flask import Flask
from flask_sqlalchemy import SQLAlchemy
import timefrom pymysql import OperationalError# 从配置文件导入数据库连接 URI
from core.config import DATABASE_URI
# 从日志模块导入日志记录器
from core.logger import Loggerlogger = Logger("info").loggerapp = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = DATABASE_URI
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
app.config['SQLALCHEMY_POOL_SIZE'] = 5  # 连接池大小
app.config['SQLALCHEMY_MAX_OVERFLOW'] = 10  # 最大溢出连接数
app.config['SQLALCHEMY_POOL_TIMEOUT'] = 30  # 连接池超时时间
app.config['SQLALCHEMY_POOL_RECYCLE'] = 3600  # 连接池回收时间# 初始化数据库连接实例
db = SQLAlchemy()
db.init_app(app)MAX_RETRIES = 3
RETRY_DELAY = 5def retry_on_failure(func):def wrapper(*args, **kwargs):for attempt in range(MAX_RETRIES):try:return func(*args, **kwargs)except OperationalError as e:if attempt < MAX_RETRIES - 1:logger.error(f"查询失败，尝试第 {attempt + 2} 次重试...")print(f"查询失败，尝试第 {attempt + 2} 次重试...")time.sleep(RETRY_DELAY)continueelse:logger.error("达到最大重试次数，查询失败。")print("达到最大重试次数，查询失败。")raisereturn Nonereturn wrapper

代码解释

1.配置 Web 框架和数据库连接库：设置数据库连接 URI、连接池大小、超时时间等参数，并且初始化数据库连接实例。
2.定义重试参数：MAX_RETRIES 代表最大重试次数，RETRY_DELAY 表示每次重试之间的间隔时间（秒）。
3. 实现 retry_on_failure 装饰器：

wrapper 函数是实际执行的函数，它会尝试调用被装饰的函数。
若在执行过程中抛出 OperationalError 异常，意味着数据库连接可能断开，此时会进行重试。
每次重试前，会记录错误日志并暂停一段时间，给数据库恢复连接的机会。
若达到最大重试次数仍失败，则抛出异常。

使用示例

以下是如何使用 retry_on_failure 装饰器的示例：

class User(db.Model):id = db.Column(db.Integer, primary_key=True)name = db.Column(db.String(100))@retry_on_failure
def get_user_by_id(user_id):return User.query.get(user_id)# 调用被装饰的函数
user = get_user_by_id(1)
if user:print(f"用户姓名: {user.name}")
else:print("未找到用户")