You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

Python脚本运行失败后如何在数天后自动重试(服务器每日重启场景)

服务器定时重启场景下Python脚本延迟重跑的最佳解决方案

核心思路是放弃依赖进程内存的sleep方案,改用磁盘持久化标记 + 系统定时任务触发的逻辑,避免服务器重启丢失调度状态。

方案1:原生系统定时任务+本地标记文件(最轻量稳定,优先推荐)

  • 第一步:改造Python脚本逻辑,执行前先校验磁盘里的执行标记,捕获异常时把下次要执行的时间戳写入本地持久化文件,执行成功则清除标记
    参考代码片段:
    import time
    import os
    
    RETRY_DELAY_DAYS = 3 # 自定义重试间隔天数
    MARK_FILE = "./next_run_mark.txt"
    
    def your_business_logic():
        # 此处写你的业务脚本逻辑
        pass
    
    if __name__ == "__main__":
        # 先校验执行标记,未到执行时间直接退出
        if os.path.exists(MARK_FILE):
            with open(MARK_FILE, "r", encoding="utf-8") as f:
                next_run_ts = float(f.read().strip())
            if time.time() < next_run_ts:
                exit(0)
            # 已到执行时间,删除标记后执行任务
            os.remove(MARK_FILE)
    
        try:
            your_business_logic()
        except Exception as e:
            # 捕获异常,写入下次执行时间标记
            next_run = time.time() + RETRY_DELAY_DAYS * 24 * 3600
            with open(MARK_FILE, "w", encoding="utf-8") as f:
                f.write(str(next_run))
    
  • 第二步:给操作系统添加定时任务,周期触发脚本运行即可:
    • Linux系统使用crontab,比如设置每天凌晨2点执行脚本,添加一行配置:0 2 * * * /usr/bin/python3 /你的脚本绝对路径/script.py
    • Windows系统使用任务计划程序,设置每天定时触发Python解释器运行脚本即可

方案2:持久化调度框架(适合多任务调度场景)

如果有多个延迟重试任务需要管理,可以直接用带持久化能力的调度框架,避免手写标记逻辑:

  • 可选APScheduler框架,搭配SQLite做任务持久化存储,把框架启动脚本加入系统开机自启项,服务器重启后框架会自动读取存储的任务状态恢复调度
    参考核心配置:
    from datetime import datetime, timedelta
    from apscheduler.schedulers.background import BackgroundScheduler
    from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore
    from apscheduler.events import EVENT_JOB_ERROR
    
    # 配置SQLite持久化任务存储
    scheduler = BackgroundScheduler(jobstores={
        "default": SQLAlchemyJobStore(url="sqlite:///scheduler_jobs.db")
    })
    
    def your_business_logic():
        # 业务逻辑
        pass
    
    # 监听任务失败事件,自动添加3天后的重试任务
    def task_fail_handler(event):
        if event.exception:
            scheduler.add_job(
                your_business_logic,
                "date",
                run_date = datetime.now() + timedelta(days=3),
                id = "retry_fail_task",
                replace_existing = True
            )
    
    scheduler.add_listener(task_fail_handler, EVENT_JOB_ERROR)
    scheduler.start()
    

内容的提问来源于stack exchange,提问作者Led

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.10.18 08:24:06