Django集成Celery Beat实现可配置定时任务详解

Django集成Celery Beat实现可配置定时任务详解

在后台功能开发中，常遇到此类需求：用户需为任务配置定时策略，使其能按预定时间执行特定操作。

需求分析

需求分析

基于上述需求，可将其拆解为以下关键步骤：

• 「某个操作」的实现
• 配置为定时任务
• 定时策略可配置
• 良好的用户体验

其中步骤1与本文主题无关，此处略过；关于定时任务的实现，前文《Celery异步任务队列》已简要提及Celery支持定时任务。

然而，Celery原生的定时任务策略配置需硬编码在代码中，启动时写入本地shelve文件，这种方式不利于长期维护和管理。

为此，Celery官方文档推荐扩展模块 django-celery-beat，该模块将定时任务配置持久化至Django数据库。程序启动后，可通过 admin 后台界面进行管理，并直接利用ORM修改配置，无需重启Celery服务，完全符合我们的预期。

尽管存在其他替代库，鉴于项目已使用Celery处理异步任务，本文继续采用 django-celery-beat 来解决此问题。

Celery定时任务采用类似 crontab 的语法，考虑到普通用户的学习成本，需提供常用预设配置（如“每工作日1点执行”），同时保留输入框以支持高级自定义，兼顾易用性与扩展性。

设计与实现

设计与实现

基本用法

定时策略（CrontabSchedule）

定时策略（CrontabSchedule）

CrontabSchedule 支持类crontab语法，包含5个配置域，分别为：

• 分钟（minute）
• 小时（hour）
• 星期几（day_of_week）
• 每月第几天（day_of_month）
• 每年第几月（month_of_year）

各配置域之间以空格分隔。

各配置域的常用语法如下：

• * : 代表该域内的所有值
• M-N : 代表M到N之间的值
• M-N/X 或 */X : 表示每X分钟、每X天等间隔
• A,B,...,Z : 枚举指定值

示例：每个工作日1点执行任务的配置为：0 1 1-5 * *

创建定时策略的代码示例如下：

from django_celery_beat.models import CrontabSchedule, PeriodicTask
>>> schedule, _ = CrontabSchedule.objects.get_or_create(
... minute='30',
... hour='*',
... day_of_week='*',
... day_of_month='*',
... month_of_year='*',
... )

定时任务

定时任务

定时任务可关联不同的策略（如crontab、interval等），创建时指定 schedule 即可。以crontab为例：

>>> import json
>>> from datetime import datetime, timedelta

>>> PeriodicTask.objects.create(
... crontab=schedule,   # we created this above.
... name='Importing contacts',  # simply describes this periodic task.
... task='proj.tasks.import_contacts', # name of task.
... args=json.dumps(['arg1', 'arg2']),
... kwargs=json.dumps({
... 'be_careful': True,
... }),
... expires=datetime.utcnow() + timedelta(seconds=30)
... )

其中 name 为任务名称，必须唯一； task 指向需执行的Celery任务函数。结合定时策略调度器，这三者构成了定时任务的核心属性。

此外，定时任务还支持其他配置，如 args / kwargs 用于传递Celery任务的入参； expires 用于设置任务的过期时间。

Django配置

Django配置

基础配置只需在 INSTALLED_APPS 中添加引用，并指定调度器即可：

settings.py

INSTALLED_APPS = [
 ...
 'django_celery_beat'
]

# 配置 celery 定时任务使用的调度器
CELERY_BEAT_SCHEDULER = 'django_celery_beat.schedulers:DatabaseScheduler'

时区问题

时区问题

在使用 django-celery-beat 过程中，常遇到两个时区相关问题：

创建的定时任务，实际触发时间与配置时间存在8小时偏差。

解决方案：

这8小时偏差通常由时区设置引起，django-celery-beat 对时区的处理可能存在固有缺陷（如有误请指正）。

需修改 settings.py 中的时区配置：

settings.py

# 设置 Django 大部分应用通用的时区
TIME_ZONE = 'Asia/Shanghai'
# 关闭 UTC
USE_TZ = False
CELERY_ENABLE_UTC = False
# 设置 django-celery-beat 真正使用的时区
CELERY_TIMEZONE = TIME_ZONE
# 使用 timezone naive 模式
DJANGO_CELERY_BEAT_TZ_AWARE = False

关于 timezone naive（无时区）与 timezone aware（有时区）模式的区别，可参考文章《Django时区详解》。

简言之，naive模式不存储时区信息，仅存储转换后的时间；aware模式则同时存储UTC时间和UTC时区信息。

根据文档说明，修改时区后，需将已执行过的定时任务的 last_run_at 重置为 None ：

python manage.py shell
>>> from django_celery_beat.models import PeriodicTask
>>> PeriodicTask.objects.all().update(last_run_at=None)

修改完成后，重启 celery beat 。

注：即使经过上述配置，仍可能遇到任务重复执行的问题，且在多次重启Celery后未再复现，因此该配置可能存在潜在问题，需结合实际测试验证。

数据库中， CrontabSchedule 的 timezone 配置始终默认为 UTC。

解决方案：

查看 CrontabSchedule 模型的源码，定位数据库中 timezone 字段的属性定义：

class CrontabSchedule(models.Model):
 ...
 timezone = timezone_field.TimeZoneField(
 default='UTC',
 verbose_name=_('Cron Timezone'),
 help_text=_(
  'Timezone to Run the Cron Schedule on. Default is UTC.'),
 )

由于创建 CrontabSchedule 实例时未指定 timezone ，因此在创建任务时，需显式添加该字段配置：

from django_celery_beat.models import CrontabSchedule
>>> schedule, _ = CrontabSchedule.objects.get_or_create(
... minute='30',
... hour='*',
... day_of_week='*',
... day_of_month='*',
... month_of_year='*',
... timezone='Asia/Shanghai'
... )

业务前后端设计

*业务前后端设计

本节内容仅供参考，不一定适用于所有场景。

前端

前端

设计前端定时任务配置项，包含一个开关、一个三选一单选组件以及一个输入框：

为提升非技术人员的操作体验，定时任务除「自定义」输入模式外，还提供「每天」与「每周」快捷选项：

• 每天：0 1 1-5 * *
• 每周：0 1 1 * *

单选框与字符串双向绑定，后端返回上述字符串之一时，前端选中“每天”或“每周”，否则选中“自定义”。

后端

后端

假设业务所需前端任务数据字段如下：

{
 "task_id": 1,
 "is_periodic_task": true,
 "periodic_task_id": 1,
 "crontab": "* * * * *"
}

ER模型如图：

返回前端的数据中，若 periodic_task 不为空，则 is_periodic_task 设为 True ，并通过 periodic_task.crontab_id 获取 CrontabSchedule 实例，将其转化为字符串返回。

需注意， CrontabSchedule 的 __str__ 方法除返回 crontab 配置外，还会返回时区等额外信息，而这些信息在前端展示中并不需要。

因此，可新建一个方法：

def get_crontab_str(contab) -> str:
 """
 获取前端配置需要的 5 项值
 :param contab: CrontabSchedule对象
 :return:
 """
 return '{0} {1} {2} {3} {4}'.format(
 cronexp(contab.minute), cronexp(contab.hour),
 cronexp(contab.day_of_week), cronexp(contab.day_of_month),
 cronexp(contab.month_of_year)
 )

在序列化时调用该方法，将结果返回给前端即可。

修改任务

修改任务

任务修改主要包含以下三种情况：

• 从定时任务改为非定时任务
• 从非定时任务改为定时任务
• 在现有定时任务基础上修改定时策略

对应流程图如下：

1:

2, 3:

图中「修改配置中的」指前端传来的修改请求中的新配置信息。

具体代码此处省略，仅提示暂停定时任务的方法：

将 PeriodicTask.objects.enabled 设置为 False/0 即可。

>>> periodic_task.enabled = False
>>> periodic_task.save()

版本说明

参考

http://docs.celeryproject.org/en/latest/userguide/periodic-tasks.html https://django-celery-beat.readthedocs.io/en/latest/ https://docs.djangoproject.com/en/2.2/topics/i18n/timezones/ https://www.jb51.net/article/166085.htm

以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持本站。