Python任务调度

小半 • 2024年3月21日下午12:04 • Python • 阅读 53

在实际的软件开发过程中，经常会遇到需要定时执行某些任务的情况，例如定时备份数据、定时发送邮件等。Python 提供了多种方式来实现任务调度，本文将介绍几种常见的任务调度方法。

一、使用 sched 模块

Python 标准库中的 sched 模块提供了一个简单的任务调度器，可以用来在指定的时间执行任务。

import sched
import time

# 创建调度器
scheduler = sched.scheduler(time.time, time.sleep)

def task():
    print("Task executed!")

# 延迟 5 秒后执行任务
scheduler.enter(5, 1, task)

# 启动调度器
scheduler.run()

二、使用 threading 模块

threading 模块可以用来创建线程，在线程中执行定时任务。

import threading
import time

def task():
    print("Task executed!")
    # 设置下一次任务执行时间间隔
    threading.Timer(5, task).start()

# 初始启动任务
task()

三、使用 schedule 库

schedule 是一个专门用于任务调度的第三方库，它提供了更加方便的调度方法和更加友好的 API。

pip install schedule

import schedule
import time

def task():
    print("Task executed!")

# 每隔 5 秒执行一次任务
schedule.every(5).seconds.do(task)

# 每隔 1 分钟执行一次任务
schedule.every().minute.do(task)

# 每隔 1 小时执行一次任务
schedule.every().hour.do(task)

# 每天的凌晨执行一次任务
schedule.every().day.at("00:00").do(task)

# 每周一的凌晨执行一次任务
schedule.every().monday.do(task)

# 每个月的第一天的凌晨执行一次任务
schedule.every().month.do(task)

# 每年的 1 月 1 日的凌晨执行一次任务
schedule.every().year.do(task)

# 取消某个任务
task = schedule.every().hour.do(task)
schedule.cancel_job(task)
while True:
    schedule.run_pending()
    time.sleep(1)

    # 不再需要定时任务时，手动停止
    if not schedule.jobs:
        break

四、使用 APScheduler 库

APScheduler 是另一个功能强大的任务调度库，它支持多种调度方式和可配置的调度器。

pip install apscheduler

from apscheduler.schedulers.blocking import BlockingScheduler
from datetime import datetime

def task():
    print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"))

# 创建调度器
scheduler = BlockingScheduler()

# 每隔 5 秒执行一次任务
scheduler.add_job(task, 'interval', seconds=5)

# 使用cron执行任务
scheduler.add_job(task, 'cron', day_of_week='1-5', hour=6, minute=30)

# 启动调度器
scheduler.start()