```
feat(app): 添加数据库重试机制提高定时任务稳定性 - 引入run_db_operation_with_retries函数处理临时性数据库错误 - 在定时任务中对订单查询操作添加重试逻辑 - 对数据库连接参数进行优化配置 - 增加适当的日志记录便于问题排查 ```
This commit is contained in:
parent
c0a0db4142
commit
43474c4395
60
app.py
60
app.py
@ -7,6 +7,7 @@ from blueprints.admin import admin_bp
|
|||||||
from blueprints.payment import payment_bp
|
from blueprints.payment import payment_bp
|
||||||
from apscheduler.schedulers.background import BackgroundScheduler
|
from apscheduler.schedulers.background import BackgroundScheduler
|
||||||
from apscheduler.triggers.interval import IntervalTrigger
|
from apscheduler.triggers.interval import IntervalTrigger
|
||||||
|
from sqlalchemy.exc import InterfaceError, OperationalError
|
||||||
import threading
|
import threading
|
||||||
import time
|
import time
|
||||||
import logging
|
import logging
|
||||||
@ -14,6 +15,33 @@ import logging
|
|||||||
# 导入模律(必需在 db.create_all() 之前导入)
|
# 导入模律(必需在 db.create_all() 之前导入)
|
||||||
import models
|
import models
|
||||||
|
|
||||||
|
TRANSIENT_DB_ERRORS = (OperationalError, InterfaceError)
|
||||||
|
DB_RETRY_FAILED = object()
|
||||||
|
|
||||||
|
def run_db_operation_with_retries(operation, retries=3, delay=2, operation_name="数据库操作", failure_value=DB_RETRY_FAILED):
|
||||||
|
"""Retry short-lived DB operations while PostgreSQL is restarting."""
|
||||||
|
last_error = None
|
||||||
|
for attempt in range(1, retries + 1):
|
||||||
|
try:
|
||||||
|
return operation()
|
||||||
|
except TRANSIENT_DB_ERRORS as e:
|
||||||
|
last_error = e
|
||||||
|
db.session.rollback()
|
||||||
|
db.engine.dispose()
|
||||||
|
if attempt < retries:
|
||||||
|
logging.warning(
|
||||||
|
"%s失败,数据库可能正在重启,%s秒后重试(%s/%s): %s",
|
||||||
|
operation_name,
|
||||||
|
delay,
|
||||||
|
attempt,
|
||||||
|
retries,
|
||||||
|
str(e),
|
||||||
|
)
|
||||||
|
time.sleep(delay)
|
||||||
|
|
||||||
|
logging.warning("%s失败,已跳过本轮: %s", operation_name, str(last_error))
|
||||||
|
return failure_value
|
||||||
|
|
||||||
# 定时任务函数
|
# 定时任务函数
|
||||||
def sync_pending_orders(app):
|
def sync_pending_orders(app):
|
||||||
"""定时任务: 检查并同步30分钟内的待支付订单"""
|
"""定时任务: 检查并同步30分钟内的待支付订单"""
|
||||||
@ -26,10 +54,15 @@ def sync_pending_orders(app):
|
|||||||
try:
|
try:
|
||||||
# 查询最还30分钟内的待支付订单
|
# 查询最还30分钟内的待支付订单
|
||||||
thirty_min_ago = get_bj_now() - timedelta(minutes=30)
|
thirty_min_ago = get_bj_now() - timedelta(minutes=30)
|
||||||
pending_orders = Order.query.filter(
|
pending_orders = run_db_operation_with_retries(
|
||||||
|
lambda: Order.query.filter(
|
||||||
Order.status == 'PENDING',
|
Order.status == 'PENDING',
|
||||||
Order.created_at >= thirty_min_ago
|
Order.created_at >= thirty_min_ago
|
||||||
).all()
|
).all(),
|
||||||
|
operation_name="定时任务查询待支付订单"
|
||||||
|
)
|
||||||
|
if pending_orders is DB_RETRY_FAILED:
|
||||||
|
return
|
||||||
|
|
||||||
if not pending_orders:
|
if not pending_orders:
|
||||||
return
|
return
|
||||||
@ -69,6 +102,15 @@ def sync_pending_orders(app):
|
|||||||
elif order_locked and order_locked.status == 'PAID':
|
elif order_locked and order_locked.status == 'PAID':
|
||||||
# 订单已经被处理,跳过
|
# 订单已经被处理,跳过
|
||||||
logging.info(f"定时任务-订单 {order.out_trade_no} 已被处理,跳过")
|
logging.info(f"定时任务-订单 {order.out_trade_no} 已被处理,跳过")
|
||||||
|
except TRANSIENT_DB_ERRORS as e:
|
||||||
|
db.session.rollback()
|
||||||
|
db.engine.dispose()
|
||||||
|
logging.warning(
|
||||||
|
"定时任务处理订单 %s 时数据库暂时不可用,跳过本轮: %s",
|
||||||
|
order.out_trade_no,
|
||||||
|
str(e),
|
||||||
|
)
|
||||||
|
return
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
# Redis lock error or other errors
|
# Redis lock error or other errors
|
||||||
if "LockError" in str(e) or "BlockingIOError" in str(e):
|
if "LockError" in str(e) or "BlockingIOError" in str(e):
|
||||||
@ -80,6 +122,10 @@ def sync_pending_orders(app):
|
|||||||
if updated_count > 0:
|
if updated_count > 0:
|
||||||
logging.info(f"定时任务完成,帮助更新了{updated_count}个订单")
|
logging.info(f"定时任务完成,帮助更新了{updated_count}个订单")
|
||||||
|
|
||||||
|
except TRANSIENT_DB_ERRORS as e:
|
||||||
|
db.session.rollback()
|
||||||
|
db.engine.dispose()
|
||||||
|
logging.warning(f"定时任务数据库暂时不可用,跳过本轮: {str(e)}")
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
logging.error(f"定时任务异常: {str(e)}", exc_info=True)
|
logging.error(f"定时任务异常: {str(e)}", exc_info=True)
|
||||||
|
|
||||||
@ -196,7 +242,15 @@ def create_app():
|
|||||||
# 自动创建数据库表
|
# 自动创建数据库表
|
||||||
with app.app_context():
|
with app.app_context():
|
||||||
print("🔧 正在检查并创建数据库表...")
|
print("🔧 正在检查并创建数据库表...")
|
||||||
db.create_all()
|
create_all_result = run_db_operation_with_retries(
|
||||||
|
db.create_all,
|
||||||
|
retries=5,
|
||||||
|
delay=3,
|
||||||
|
operation_name="启动时创建数据库表"
|
||||||
|
)
|
||||||
|
if create_all_result is DB_RETRY_FAILED:
|
||||||
|
print("⚠️ 数据库暂时不可用,跳过启动时建表检查")
|
||||||
|
else:
|
||||||
print("✅ 数据库表已就绪")
|
print("✅ 数据库表已就绪")
|
||||||
|
|
||||||
# 创建并启动定时任务调度器
|
# 创建并启动定时任务调度器
|
||||||
|
|||||||
@ -10,7 +10,10 @@ class Config:
|
|||||||
SQLALCHEMY_ENGINE_OPTIONS = {
|
SQLALCHEMY_ENGINE_OPTIONS = {
|
||||||
"pool_pre_ping": True,
|
"pool_pre_ping": True,
|
||||||
"pool_recycle": 1800,
|
"pool_recycle": 1800,
|
||||||
"pool_timeout": 30
|
"pool_timeout": 30,
|
||||||
|
"connect_args": {
|
||||||
|
"connect_timeout": 5
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
# Redis 配置
|
# Redis 配置
|
||||||
|
|||||||
Loading…
Reference in New Issue
Block a user