前言
在基于 Python 的 Web 应用开发里，数据库连接是极为关键的一环。不过，像网络波动、数据库服务器维护这类因素，都可能造成数据库长时间断连，进而影响应用的正常运作。本文将详细介绍怎样运用 retry_on_failure 装饰器来解决数据库长时间断连的难题
一 问题背景
在实际开发场景中，应用和数据库之间的连接可能会由于各种缘由中断（长时间系统无人访问，再次访问，数据库连接超时）。当应用尝试执行数据库操作时，要是连接已经断开，就会抛出异常，致使请求失败。从用户角度来看，这或许表现为页面加载失败或者操作无响应，极大地影响用户体验。所以，我们需要一种机制来应对数据库断连的状况，尽可能恢复连接并重新执行操作。
二 程序层面解决方案：retry_on_failure 装饰器
装饰器是 Python 中强大的语法糖，它能够在不修改原有函数代码的基础上，为函数增添额外功能。我们可以借助装饰器实现数据库操作的重试机制，当数据库连接断开时，自动重试一定次数，提升操作成功的几率。
实现代码
下面是 retry_on_failure 装饰器的实现代码，同时包含了 Web 框架和数据库连接库的基本配置：
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
import time
from pymysql import OperationalError
# 从配置文件导入数据库连接 URI
from core.config import DATABASE_URI
# 从日志模块导入日志记录器
from core.logger import Logger
logger = Logger("info").logger
app = 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 = 5
def 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)
continue
else:
logger.error("达到最大重试次数，查询失败。")
print("达到最大重试次数，查询失败。")
raise
return None
return 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("未找到用户")
三 总结
通过运用 retry_on_failure 装饰器，能够有效处理数据库长时间断连的问题，增强应用的健壮性和稳定性。当数据库连接断开时，应用会自动重试一定次数，减少因网络波动或数据库临时故障导致的请求失败。在实际应用中，可以依据具体情况调整最大重试次数和重试间隔时间，以实现最佳效果。
- 注意事项
尽管重试机制能提高应用的容错能力，但如果数据库长时间无法恢复连接，最终还是会失败。所以，需要及时排查数据库故障并进行修复。
频繁的重试可能会加重服务器负担，建议合理设置重试次数和间隔时间。