使用Python编写定时任务

定时任务是自动化操作的重要组成部分，广泛应用于数据处理、系统监控、定时备份等多个领域。Python作为一种强大且灵活的编程语言，提供了多种方式来实现定时任务。本文将介绍如何使用Python编写一个简单的定时任务，并给出具体的代码示例。

1. 使用 time.sleep()

最简单的方法是使用 time.sleep() 函数来实现定时任务。这种方法适用于简单的场景，但对于复杂的任务调度并不推荐。

import time def my_task():    print("定时任务执行中...") while True:    my_task()    time.sleep(60)  # 每60秒执行一次

2. 使用 schedule 库

schedule 是一个轻量级的定时任务调度库，非常适合用于简单的定时任务。首先需要安装 schedule 库：

pip install schedule

然后可以使用以下代码来创建一个定时任务：

import schedule import time def my_task():    print("定时任务执行中...") # 每隔1分钟执行一次 schedule.every(1).minutes.do(my_task) # 每天的特定时间执行 schedule.every().day.at("10:30").do(my_task) # 每小时执行一次 schedule.every().hour.do(my_task) # 每周一执行一次 schedule.every().monday.do(my_task) # 每周五15:30执行一次 schedule.every().friday.at("15:30").do(my_task) while True:    schedule.run_pending()    time.sleep(1)

3. 使用 APScheduler 库

APScheduler 是一个功能更强大的定时任务调度库，支持多种调度策略和持久化存储。首先需要安装 APScheduler 库：

pip install apscheduler

然后可以使用以下代码来创建一个定时任务：

from apscheduler.schedulers.blocking import BlockingScheduler def my_task():    print("定时任务执行中...") # 创建调度器 scheduler = BlockingScheduler() # 每隔1分钟执行一次 scheduler.add_job(my_task, 'interval', minutes=1) # 每天的特定时间执行 scheduler.add_job(my_task, 'cron', hour=10, minute=30) # 每小时执行一次 scheduler.add_job(my_task, 'interval', hours=1) # 每周一执行一次 scheduler.add_job(my_task, 'cron', day_of_week='mon') # 每周五15:30执行一次 scheduler.add_job(my_task, 'cron', day_of_week='fri', hour=15, minute=30) # 启动调度器 scheduler.start()

4. 使用 Celery 和 Redis

对于更复杂的应用场景，可以使用 Celery 结合 Redis 或其他消息队列来实现分布式定时任务。首先需要安装 Celery 和 Redis：

pip install celery redis

然后可以使用以下代码来配置和运行定时任务：

1. 配置 Celery

    

   在项目根目录下创建一个 celery.py 文件：

from celery import Celery app = Celery('tasks', broker='redis://localhost:6379/0', backend='redis://localhost:6379/0') @app.task def my_task():    print("定时任务执行中...")

配置定时任务

在项目根目录下创建一个 beat_schedule.py 文件：

from celery.schedules import crontab app.conf.beat_schedule = {    'run-every-minute': {        'task': 'tasks.my_task',        'schedule': crontab(minute='*/1'),    },    'run-daily-at-midnight': {        'task': 'tasks.my_task',        'schedule': crontab(hour=0, minute=0),    }, }

启动 Celery Worker 和 Beat

打开两个终端窗口，分别运行以下命令：

celery -A celery worker --loglevel=info celery -A celery beat --loglevel=info

time.sleep() 适合简单的场景，schedule 库适合轻量级的任务调度，而 APScheduler 和 Celery 则适合更复杂和高性能的场景。