在之前的文章中,我们通过 Django-Celery-Beat: 实际例子利用 PeriodicTask 设置定时任务 讨论了如何管理 使用 IntervalSchedule 的定时任务

然而,在实际项目中,仅使用重复执行(interval)方式往往不够

  • 当需要在特定时间执行任务时
  • 当需要在特定日期和时间只执行一次

在这些情况下,使用 Crontab 或 Clocked 调度方式可以设置更精确的定时任务。🚀

📌 不熟悉 Crontab 概念吗?

➡️ 正确理解 Linux `crontab`: 开发者爱的自动调度技巧

先阅读以上文章会更容易理解!😊

Django Celery-Beat 可以通过 Crontab 和 Clocked 进行任务调度

1️⃣ 使用 `CrontabSchedule` 进行特定时间的调度

Crontab 是一种在 Linux 上使用的定时任务方式,可以设置在特定时间(如:秒/分钟/小时/星期/日期)执行

在 Django-Celery-Beat 中,也可以使用 `CrontabSchedule` 以相同的方式设置定时任务。

📌 示例:每天上午 9:30 执行的定时任务

from django_celery_beat.models import PeriodicTask, CrontabSchedule
import json

# 1️⃣ 创建 CrontabSchedule(每天 09:30 执行)
schedule, created = CrontabSchedule.objects.get_or_create(
    minute="30",
    hour="9",
    day_of_week="*",
    day_of_month="*",
    month_of_year="*"
)

# 2️⃣ 创建或更新 PeriodicTask
task, created = PeriodicTask.objects.update_or_create(
    name="daily_report_task",
    defaults={
        "crontab": schedule,  # 应用 CrontabSchedule
        "task": "myapp.tasks.generate_daily_report",
        "args": json.dumps([]),
        "enabled": True,
    }
)

✅ 以上设置将安排每天上午 9:30 执行!🎯

✅ `CrontabSchedule` 字段说明

字段名 说明 示例
`minute` 设置执行的分钟 "0", "30", "*/10"
`hour` 设置执行的小时 "0", "9", "*/6"
`day_of_week` 设置执行的星期(0=星期日) "*"(每天),`"1,3,5"`(周一/周三/周五)
`day_of_month` 设置执行的日期 "*"(每天),`"1,15"`(每月 1 号和 15 号)
`month_of_year` 设置执行的月份 "*"(每月),`"1,6,12"`(1 月、6 月、12 月)

2️⃣ 使用 `ClockedSchedule` 进行特定时间执行一次的调度

`ClockedSchedule` 用于设置在特定日期和时间只执行一次的定时任务。

📌 示例:在特定日期(2025 年 2 月 10 日 15:00)只执行一次的定时任务

from django_celery_beat.models import PeriodicTask, ClockedSchedule
import json
from datetime import datetime

# 1️⃣ 创建 ClockedSchedule(在 2025 年 2 月 10 日 15:00 执行)
scheduled_time = datetime(2025, 2, 10, 15, 0)  # 需要使用 datetime 对象!
schedule, created = ClockedSchedule.objects.get_or_create(
    clocked_time=scheduled_time
)

# 2️⃣ 创建 PeriodicTask(设置 one_off=True!)
task, created = PeriodicTask.objects.update_or_create(
    name="one_time_task_20250210",
    defaults={
        "clocked": schedule,  # 应用 ClockedSchedule
        "task": "myapp.tasks.one_time_email",
        "args": json.dumps([]),
        "enabled": True,
        "one_off": True,  # 执行后自动禁用
    }
)

✅ 以上设置将在 2025 年 2 月 10 日 15:00 只执行一次,然后 自动禁用

  • 📌 创建 ClockedSchedule 模型实例时,clocked_time 字段必须输入 datetime 对象。
  • 📌 使用字符串或其他数据类型可能会导致错误。

📌 ClockedSchedule 和 one_off 选项的关系是什么?

ClockedSchedule 中,创建只执行一次的任务时,要设置 one_off=True
这项设置如何与 enabled 字段相互作用可能是你想知道的。

📌 如果你想更深入了解这两个字段(clocked_timeone_off)之间的关系?

➡️ Celery Beat 中 one-off 和 enabled 的关系:正确理解和使用方法

在上面的文章中,one_off=True 是如何确切工作的,以及 enabled 与其有什么不同都进行了详细讨论,欢迎参考!😊

🎯 结论:通过 CrontabClocked 可以实现更精细的定时任务!

  • 使用 CrontabSchedule 可以安排 定期但在特定时间执行的任务
  • 使用 ClockedSchedule 可以安排 在特定时刻只执行一次的任务
  • 创建 ClockedSchedule 时,clocked_time 字段必须输入 datetime 对象!
  • ClockedSchedule 和 one_off=True 设置之间的关系将在另一个帖子中深入探讨,敬请关注!

如果你想知道更深入的 Celery 使用方法,

请在下面的 '相关帖子' 列表中查看,或者在右上角的搜索框中搜索 `"Celery"`!🚀 😊