一文带你深入了解Python中的自动化任务调度

import schedule
import time # 用于暂停程序，让调度器有机会运行
import os
import datetime

# 假设这个函数就是你之前写的清理函数，这里为了演示简化为打印日志
def perform_cleanup_job(folder_path, days_threshold):
    """
    模拟一个文件清理任务。
    在实际应用中，这里会调用你之前编写的 auto_clean_folder_once 函数。
    """
    print(f"⏰ [{datetime.datetime.now().strftime('%H:%M:%S')}] 正在执行定时清理任务...")
    print(f"   - 清理目录: {folder_path}")
    print(f"   - 删除 {days_threshold} 天前的过期文件。")
    # 在实际代码中，这里会调用 auto_clean_folder_once(folder_path, days_threshold)
    print("   ✨ 本次清理任务完成！")

if __name__ == "__main__":
    # 请替换为你要定时清理的文件夹路径
    target_folder_for_cleanup = os.path.expanduser("~/Desktop/每日自动清理") 
    os.makedirs(target_folder_for_cleanup, exist_ok=True) # 确保文件夹存在

    print("🚀 正在启动Python自动化任务调度器 (schedule库演示)...")
    print("--- 请注意：此脚本需要持续运行才能保持调度！ ---")

    # 示例1：每隔5秒执行一次任务
    schedule.every(5).seconds.do(perform_cleanup_job, target_folder_for_cleanup, 7) # 模拟清理7天前文件
    print(f"任务已设置：每5秒执行一次清理任务，目标文件夹：{target_folder_for_cleanup}")

    # 示例2：每隔1分钟执行一次
    # schedule.every(1).minutes.do(perform_cleanup_job, target_folder_for_cleanup, 14)
    # print("任务已设置：每1分钟执行一次清理任务。")

    # 示例3：每天的特定时间执行任务 (例如每天凌晨1点)
    # schedule.every().day.at("01:00").do(perform_cleanup_job, target_folder_for_cleanup, 30)
    # print("任务已设置：每天凌晨01:00执行清理任务。")

    # 示例4：每周三的下午14:30执行
    # schedule.every().wednesday.at("14:30").do(perform_cleanup_job, target_folder_for_cleanup, 60)
    # print("任务已设置：每周三14:30执行清理任务。")

    # 主循环：持续检查是否有待执行的任务
    while True:
        schedule.run_pending() # 运行所有待执行的任务
        time.sleep(1) # 等待1秒，避免CPU占用过高

步骤：

安装schedule：在VS Code终端运行 pip install schedule。

修改代码路径：复制上方代码到VS Code，保存为schedule_basic_demo.py。修改 target_folder_for_cleanup。

运行：在VS Code终端运行 python schedule_basic_demo.py。

展示：

schedule库虽然简单易用，但它需要脚本持续运行在内存中，并且没有内置的持久化、多线程、作业重试等功能。

对于更复杂、更健壮的Python自动化任务调度需求，你需要APScheduler（Advanced Python Scheduler）。它是一个功能全面的调度库，能够应对更严苛的生产环境。

作用： APScheduler提供了一个灵活的调度器，可以脱离主程序线程运行，支持多种存储（内存、数据库）和执行方式。

安装APScheduler：

pip install APScheduler

常用组件：

Scheduler (调度器)：核心组件，负责管理和触发作业（Job）。

Job Store (作业存储)：存储作业的地方，可以是内存、数据库（SQLite, MongoDB, PostgreSQL等）。这使得任务可以持久化，即使程序重启也不会丢失。

Executor (执行器)：定义作业如何被执行（如线程池、进程池）。

Trigger (触发器)：定义作业何时运行（如日期、间隔、Cron表达式）。

场景：你需要一个更可靠的定时任务，或者想在特定的日期执行一次性任务，或更灵活地使用Cron表达式。

方案： APScheduler支持多种触发器，让你对任务执行时间有更精准的控制。

代码：

from apscheduler.schedulers.background import BackgroundScheduler # 后台调度器
from apscheduler.triggers.interval import IntervalTrigger # 间隔触发器
from apscheduler.triggers.date import DateTrigger # 日期触发器
from apscheduler.triggers.cron import CronTrigger # Cron表达式触发器
import os
import datetime
import time

def perform_data_sync_job():
    """
    模拟一个数据同步任务。
    实际中，这里可以调用你的其他自动化脚本。
    """
    print(f"🔄 [{datetime.datetime.now().strftime('%H:%M:%S')}] 正在执行数据同步任务...")
    # 模拟任务耗时
    time.sleep(2)
    print("   ✨ 数据同步任务完成！")

def perform_weekly_report_job():
    """模拟一个每周生成报告的任务"""
    print(f"📊 [{datetime.datetime.now().strftime('%H:%M:%S')}] 正在生成每周报告...")
    time.sleep(3)
    print("   ✨ 每周报告生成完成！")

if __name__ == "__main__":
    scheduler = BackgroundScheduler() # 创建一个后台调度器实例
    
    print("🚀 正在启动APScheduler任务调度器...")

    # 1. 间隔触发器 (IntervalTrigger)：每隔5秒执行一次
    scheduler.add_job(
        perform_data_sync_job, 
        IntervalTrigger(seconds=5), 
        id='data_sync_job' # 为作业指定唯一ID
    )
    print("任务：'data_sync_job' 已设置，每5秒执行一次。")

    # 2. 日期触发器 (DateTrigger)：在指定日期和时间执行一次性任务
    # run_date = datetime.datetime.now() + datetime.timedelta(seconds=10)
    # scheduler.add_job(
    #     perform_weekly_report_job, 
    #     DateTrigger(run_date=run_date), 
    #     id='one_time_report', 
    #     name='一次性报告任务'
    # )
    # print(f"任务：'one_time_report' 已设置，将在 {run_date.strftime('%H:%M:%S')} 执行。")

    # 3. Cron触发器 (CronTrigger)：使用Cron表达式进行复杂调度
    # 每天凌晨2点0分执行
    scheduler.add_job(
        perform_weekly_report_job, 
        CronTrigger(hour=2, minute=0), 
        id='daily_cleanup_cron', 
        name='每日凌晨清理任务'
    )
    print("任务：'daily_cleanup_cron' 已设置，每天凌晨2点执行。")

    # 启动调度器
    scheduler.start()
    print("调度器已启动。程序将持续运行以保持调度。按 Ctrl+C 停止。")

    # 保持主线程运行，以便后台调度器能够工作
    try:
        while True:
            time.sleep(1)
    except (KeyboardInterrupt, SystemExit):
        scheduler.shutdown() # 安全关闭调度器
        print("调度器已停止。")

步骤：

安装APScheduler： pip install APScheduler。

修改代码：复制上方代码到VS Code，保存为apscheduler_demo.py。

运行：在VS Code终端运行 python apscheduler_demo.py。

效果展示：

对于生产环境的Python自动化任务调度，任务的健壮性和可靠性至关重要。

APScheduler提供了：

异常处理：你可以为作业注册监听器，捕获任务执行中的异常，进行日志记录或邮件通知，确保即使任务失败也能及时知晓。

持久化存储：使用SQLAlchemyJobStore等，将作业信息存储到数据库中。这样，即使程序崩溃或重启，APScheduler也能从数据库中恢复未完成或计划中的任务，实现无人值守自动运行**的真正持久性。

代码：

from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore # 用于持久化
from apscheduler.triggers.interval import IntervalTrigger
import os
import datetime
import time

# 定义一个会失败的模拟任务
def failing_job():
    print(f"🔴 [{datetime.datetime.now().strftime('%H:%M:%S')}] 尝试执行一个可能失败的任务...")
    if datetime.datetime.now().second % 10 == 0: # 每10秒模拟失败一次
        raise ValueError("模拟任务执行失败！")
    print("   ✅ 任务成功完成！")

# 定义一个错误监听器
def my_listener(event):
    if event.exception:
        print(f"🔥 作业 '{event.job_id}' 执行失败：{event.exception}")
    else:
        print(f"✨ 作业 '{event.job_id}' 正常完成。")

if __name__ == "__main__":
    # 配置作业存储：这里使用SQLite作为示例数据库
    jobstores = {
        'default': SQLAlchemyJobStore(url='sqlite:///jobs.sqlite')
    }
    # 配置执行器 (默认是ThreadPoolExecutor)
    executors = {
        'default': {'type': 'threadpool', 'max_workers': 20}
    }

    scheduler = BackgroundScheduler(jobstores=jobstores, executors=executors)
    
    print("🚀 正在启动APScheduler调度器 (持久化与异常处理演示)...")

    # 添加一个每5秒执行的任务
    scheduler.add_job(
        failing_job, 
        IntervalTrigger(seconds=5), 
        id='resilient_task',
        name='弹性任务'
    )
    print("任务：'resilient_task' 已设置，每5秒执行一次。")

    # 注册错误监听器
    from apscheduler.events import EVENT_JOB_ERROR, EVENT_JOB_EXECUTED
    scheduler.add_listener(my_listener, EVENT_JOB_ERROR | EVENT_JOB_EXECUTED)

    # 启动调度器
    scheduler.start()
    print("调度器已启动。可以尝试 Ctrl+C 中断程序，再重新运行，任务会从上次中断的地方继续。")

    try:
        while True:
            time.sleep(1)
    except (KeyboardInterrupt, SystemExit):
        scheduler.shutdown()
        print("调度器已停止。")

步骤：

安装库： pip install APScheduler SQLAlchemy。

修改代码：复制上方代码到VS Code，保存为apscheduler_advanced.py。

运行：运行 python apscheduler_advanced.py。观察日志。尝试在任务执行中按 Ctrl+C 中断程序，然后重新运行脚本，看任务是否能继续。

展示：

通过本篇文章，你已经掌握了Python自动化任务调度的各项核心魔法，亲手打造了一个能够让你的脚本无人值守自动运行的Python自动化工具集！

我们深入学习了schedule库和APScheduler，它们是实现Python定时任务的强大工具，

实现了：

schedule的轻量级调度：快速实现简单的、基于时间间隔或特定时间点的定时执行。

APScheduler的高级调度：支持多种触发器（间隔、日期、Cron），实现后台运行、持久化存储和异常处理，让你的任务更加健壮可靠。

现在，你不再需要为手动启动自动化脚本而烦恼。

通过本篇文章，你已经掌握了Python自动化任务调度的终极奥秘，为你的办公自动化之旅又增添了一个重量级技能！你学会了如何利用schedule和APScheduler，高效地让Python脚本无人值守自动运行。

表格：schedule vs. APScheduler对比

特性 schedule APScheduler

易用性极简API，快速上手功能全面，API稍复杂，但更灵活

运行方式依赖主程序while True循环，非真正后台可在后台线程/进程运行，不阻塞主程序

持久化 ❌ 不支持 ✅ 支持 (可存数据库，不怕重启)

触发器简单时间间隔/定点/周几间隔、日期、Cron表达式等多种

错误处理需手动捕获并处理内置监听器，可定制错误事件处理

适用场景简单个人脚本，短期运行生产环境，复杂调度，需健壮性任务

图注：Python两大任务调度库schedule与APScheduler对比，选择适合你需求的工具！

现在，你的Python自动化工具已经拥有了“终极自由”！它可以独立运行，无需你的关注，默默地提升你的效率提升和电脑优化。

到此这篇关于一文带你深入了解Python中的自动化任务调度的文章就介绍到这了,更多相关Python任务调度内容请搜索风君子博客以前的文章或继续浏览下面的相关文章希望大家以后多多支持风君子博客！

您可能感兴趣的文章:

Python中轻量级定时任务调度库schedule使用指南

基于Python schedule的任务调度详解

Python轻量级定时任务调度APScheduler的使用

python周期任务调度工具Schedule使用详解

Python使用定时调度任务的方式

Python任务调度模块APScheduler使用

Python任务调度利器之APScheduler详解

python任务调度实例分析

华夏福临门盛世加强版怎么样？产品责任是重中之重

国寿福庆典版值得买吗？附投保建议

理财中低风险是几级？中低风险理财会亏吗

瑞泰泰安心防癌险优缺点是什么？这份分析不容错过

2020最新自助免费申请Office365教育版，免费5TOneDrive云盘详细图文教程

MacBook怎么绕过BootCamp安装Win10双系统？

国内知名的前端博客

域名为什么会被墙？如何检测域名是否被墙？域名被墙如何处理？

最新的windows xp sp3产品密匙 xp sp3序列号 xp sp3激活码

特性	schedule	APScheduler
易用性	极简API，快速上手	功能全面，API稍复杂，但更灵活
运行方式	依赖主程序while True循环，非真正后台	可在后台线程/进程运行，不阻塞主程序
持久化	❌ 不支持	✅ 支持 (可存数据库，不怕重启)
触发器	简单时间间隔/定点/周几	间隔、日期、Cron表达式等多种
错误处理	需手动捕获并处理	内置监听器，可定制错误事件处理
适用场景	简单个人脚本，短期运行	生产环境，复杂调度，需健壮性任务
图注：Python两大任务调度库schedule与APScheduler对比，选择适合你需求的工具！

一文带你深入了解Python中的自动化任务调度

目录

在办公自动化中，仅仅编写好脚本是第一步。如何让这些脚本在需要的时候自动运行，无需人工干预，是实现真正无人值守自动化的关键。

更多文章

站内搜索

标签云

热门文章

友情链接