任务调度

在实际的自动化流程中,许多任务需要在特定时间或以固定的间隔自动执行。任务调度正是为了解决“如何让脚本在预定时间自动运行”的问题。

例如,每天晚上定时备份文件,可以保证数据安全;或者定时检查系统状态,确保服务器稳定运行。

Python 提供了多种方式来实现任务调度,其中常用的有:

  • schedule 模块:适用于简单的定时任务调度,语法简单、易于上手。
  • APScheduler 模块:功能更加强大,支持多种调度方式(例如 cron 表达式),适用于更复杂的任务调度场景。

任务调度可以帮助我们:

  • 自动化执行:无需人工干预,任务按预定计划自动运行,提高工作效率。
  • 数据安全:例如定时备份数据,可以防止因意外情况而导致的数据丢失。
  • 系统监控:定时检测系统状态,及时发现并处理异常问题。
  • 降低运维成本:自动化的任务调度减少了日常手动操作,降低了人力成本。

通过学习任务调度技术,您不仅能提高自动化脚本的执行效率,还能为很多实际业务场景(如数据备份、日志清理、定时报告生成等)提供可靠的技术支撑。

使用 schedule 模块实现定时任务

`schedule` 模块非常轻量,适合于编写简单的定时任务。其核心思想是定义任务函数,并使用 `schedule.every().day.at(“HH:MM”)` 指定任务的运行时间,再在循环中不断检查是否有任务需要运行。

下面的示例代码展示了如何使用 `schedule` 模块每天晚上 23:00 自动执行文件备份任务。该任务将指定目录中的所有文件备份到备份目录中。

import os
import shutil
import schedule
import time

def backup_files():
    source_dir = "data"     # 原始文件所在目录
    backup_dir = "backup"   # 备份文件存放目录

    # 如果备份目录不存在,自动创建
    if not os.path.exists(backup_dir):
        os.makedirs(backup_dir)
        print(f"创建备份目录: {backup_dir}")

    try:
        # 遍历源目录中的所有文件,执行备份
        for filename in os.listdir(source_dir):
            source_file = os.path.join(source_dir, filename)
            backup_file = os.path.join(backup_dir, filename)
            shutil.copy(source_file, backup_file)
        print("备份完成!")
    except Exception as e:
        print(f"备份过程中出错:{e}")

# 定义每天23:00执行备份任务
schedule.every().day.at("23:00").do(backup_files)

# 持续检查并执行调度任务
while True:
    schedule.run_pending()
    time.sleep(1)

代码说明

  • 任务函数 backup_files:负责备份指定目录下的所有文件。
  • 调度设定schedule.every().day.at("23:00").do(backup_files) 表示每天 23:00 触发备份任务。
  • 循环执行:在无限循环中调用 schedule.run_pending() 来检查并执行任务。

提示词示例:

提示词:

请编写一个Python脚本,使用schedule模块每天晚上23:00自动备份文件夹"data"中的所有文件到文件夹"backup"中,如果"backup"不存在则自动创建。如果在备份过程中出现错误,请捕获异常并输出详细错误信息。

使用 APScheduler 实现定时任务

APScheduler 是一个功能强大的 Python 任务调度库,支持多种调度方式(如 cron 表达式、固定间隔调度等),适合于更复杂和灵活的任务调度需求。相比于 schedule,APScheduler 能够更精细地控制任务的执行时间和频率。

下面的示例代码展示了如何使用 APScheduler 实现每天晚上 23:00 执行备份任务。我们使用 BlockingScheduler 来创建调度器,并添加一个基于 cron 表达式的任务。

from apscheduler.schedulers.blocking import BlockingScheduler
import os
import shutil

def backup_files():
    source_dir = "data"     # 原始文件所在目录
    backup_dir = "backup"   # 备份文件存放目录

    # 如果备份目录不存在,则创建
    if not os.path.exists(backup_dir):
        os.makedirs(backup_dir)
        print(f"创建备份目录: {backup_dir}")

    try:
        # 遍历源目录中的所有文件,执行备份操作
        for filename in os.listdir(source_dir):
            source_file = os.path.join(source_dir, filename)
            backup_file = os.path.join(backup_dir, filename)
            shutil.copy(source_file, backup_file)
        print("备份完成!")
    except Exception as e:
        print(f"备份过程中出错:{e}")

# 创建一个阻塞式调度器
scheduler = BlockingScheduler()

# 添加任务:每天23:00执行备份任务(使用cron表达式)
scheduler.add_job(backup_files, 'cron', hour=23, minute=0)

# 启动调度器
scheduler.start()

代码说明

  • 任务函数 backup_files:与 schedule 示例中相同,用于执行文件备份操作。
  • 调度器设置:使用 BlockingScheduler 创建调度器,并通过 scheduler.add_job 添加一个基于 cron 表达式的任务,该任务每天 23:00 执行。
  • 启动调度器:调用 scheduler.start() 后,调度器会阻塞当前线程,按设定时间执行任务。

提示词示例:

提示词:

请编写一个Python脚本,使用APScheduler实现每天23:00自动备份文件夹"data"中的所有文件到文件夹"backup"中,并在执行前检测文件夹是否存在,不存在则自动创建。同时在备份过程中如果发生异常,请记录错误并输出提示信息。

练习:

1、请设计提示词来修改现有的备份脚本,在备份成功后记录备份成功的详细日志(包括时间、文件数量等)。

2、思考题:请思考 schedule 模块和 APScheduler 模块各自的优缺点,以及在什么场景下更适合使用其中一种调度方法。

AI 助教

提示:您可在此提出学习中遇到的问题。回答由 AI 生成,可能存在错误,请注意甄别。