Celery 周期任务与调度器配置

本文对应 Celery 5.6 稳定版文档。

简介

celery beat 是调度器，定期触发任务，再由集群中可用的工作节点执行。

调度条目默认来自 beat_schedule，也可以使用自定义存储，例如将条目保存在 SQL 数据库中。

同一份调度计划在任何时刻都只能有一个调度器运行，否则会重复发送任务。集中调度意味着无需同步计划，服务也可以不借助锁运行。

时区

周期任务默认使用 UTC，可通过 timezone 修改。例如：

timezone = 'Europe/London'

将设置直接应用到 app，例如 app.conf.timezone = 'Europe/London'，或放入通过 app.config_from_object 加载的配置模块。更多选项见 配置文档。

默认调度器将计划存入 celerybeat-schedule 文件，能自动发现时区变化并重置计划。其他调度器，例如 Django 数据库调度器，未必具备这一能力，需要手动重置。

Django 用户

Celery 推荐使用并兼容 Django 1.4 引入的 USE_TZ。

Django 用户默认使用 TIME_ZONE 指定的时区，也可以通过 timezone 单独为 Celery 设置。

数据库调度器不会在时区设置变化后自动重置，需要手动执行：

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

原文说明 Django-Celery 仅支持 Celery 4.0 及以下，而 Celery 4.0 及以上应使用以下方式：

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

调度条目

要定期调用任务，需要向 beat 调度列表添加条目：

from celery import Celery
from celery.schedules import crontab

app = Celery()

@app.on_after_configure.connect
def setup_periodic_tasks(sender: Celery, **kwargs):
    # Calls test('hello') every 10 seconds.
    sender.add_periodic_task(10.0, test.s('hello'), name='add every 10')

    # Calls test('hello') every 30 seconds.
    # It uses the same signature of previous task, an explicit name is
    # defined to avoid this task replacing the previous one defined.
    sender.add_periodic_task(30.0, test.s('hello'), name='add every 30')

    # Calls test('world') every 30 seconds
    sender.add_periodic_task(30.0, test.s('world'), expires=10)

    # Executes every Monday morning at 7:30 a.m.
    sender.add_periodic_task(
        crontab(hour=7, minute=30, day_of_week=1),
        test.s('Happy Mondays!'),
    )

@app.task
def test(arg):
    print(arg)

@app.task
def add(x, y):
    z = x + y
    print(z)

在 on_after_configure 处理器中设置，可以避免调用 test.s() 时在模块级别求值 app。注意，该信号在 app 配置完成后发出；定义 app 的模块之外的任务，例如通过 autodiscover_tasks() 找到的 tasks.py，应使用更晚的信号，例如 on_after_finalize。

add_periodic_task() 在底层把条目加入 beat_schedule。也可以直接手动配置，例如每 30 秒运行一次 tasks.add：

app.conf.beat_schedule = {
    'add-every-30-seconds': {
        'task': 'tasks.add',
        'schedule': 30.0,
        'args': (16, 16)
    },
}
app.conf.timezone = 'UTC'

注意：设置可以直接赋给 app，也可以保存在单独配置模块中。单元素 args 元组需要尾随逗号，因为形成元组的是逗号，而不是括号。

使用 timedelta 指定 30 秒间隔时，第一次任务在 beat 启动 30 秒后发送，之后在上次运行 30 秒后再次发送。

还支持类似 crontab 的计划，见下文。与 cron 一样，如果上一次还没完成，下一次就开始，任务会重叠。若这会造成问题，应采用锁策略，保证同一时间只有一个实例运行，参见 确保任务一次只执行一个实例。

可用字段

• task：待执行任务的名称。名称规则见用户指南“Names”。虽然默认命名看起来像导入路径，但任务名并不等于导入路径。
• schedule：执行频率。可以是整数秒数、timedelta 或 crontab，也可以扩展 schedule 接口定义自定义类型。
• args：位置参数，列表或元组。
• kwargs：关键字参数，字典。
• options：执行选项字典，可以包含 apply_async() 支持的参数，例如 exchange、routing_key、expires。
• relative：为 true 时，timedelta 调度按时钟对齐，根据周期大小舍入到最近的秒、分钟、小时或天。默认 false，不舍入，而是相对于 beat 启动时间。

Crontab 调度

需要更精确地指定执行时间，例如某个星期几或一天中的某个时刻，可以使用 crontab：

from celery.schedules import crontab

app.conf.beat_schedule = {
    # Executes every Monday morning at 7:30 a.m.
    'add-every-monday-morning': {
        'task': 'tasks.add',
        'schedule': crontab(hour=7, minute=30, day_of_week=1),
        'args': (16, 16),
    },
}

表达式语法很灵活，下面是一些示例：

表达式	含义
crontab()	每分钟执行。
crontab(minute=0, hour=0)	每天午夜执行。
crontab(minute=0, hour='*/3')	每三小时执行：0、3、6、9、12、15、18、21 点。
crontab(minute=0,hour='0,3,6,9,12,15,18,21')	与上一项相同。
crontab(minute='*/15')	每 15 分钟执行。
crontab(day_of_week='sunday')	星期日的每一分钟执行。
crontab(minute='*',hour='*', day_of_week='sun')	与上一项相同。
crontab(minute='*/10',hour='3,17,22', day_of_week='thu,fri')	星期四、星期五的 3–4 点、17–18 点、22–23 点之间，每 10 分钟执行。
crontab(minute=0, hour='*/2,*/3')	每个偶数小时，以及每个能被 3 整除的小时执行；即除了 1、5、7、11、13、17、19、23 点以外的整点。
crontab(minute=0, hour='*/5')	小时数能被 5 整除时执行。例如 15 点会触发，而 17 点不会。
crontab(minute=0, hour='*/3,8-17')	能被 3 整除的整点，以及 8–17 点办公时段的每个整点执行。
crontab(0, 0, day_of_month='2')	每月 2 日执行。
crontab(0, 0,day_of_month='2-30/2')	每个偶数日期执行。
crontab(0, 0,day_of_month='1-7,15-21')	每月第一周和第三周执行。
crontab(0, 0, day_of_month='11',month_of_year='5')	每年 5 月 11 日执行。
crontab(0, 0,month_of_year='*/3')	每季度第一个月的每天执行。

更多说明见 celery.schedules.crontab。

太阳事件调度

如果任务应随日出、日落、黎明或黄昏执行，可以使用 solar：

from celery.schedules import solar

app.conf.beat_schedule = {
    # Executes at sunset in Melbourne
    'add-at-melbourne-sunset': {
        'task': 'tasks.add',
        'schedule': solar('sunset', -37.81753, 144.96715),
        'args': (16, 16),
    },
}

参数为 solar(event, latitude, longitude)。注意经纬度符号：纬度正值代表北，负值代表南；经度正值代表东，负值代表西。

支持以下事件：

事件	含义
dawn_astronomical	天文晨光开始，天空不再完全黑暗；太阳在地平线下 18 度。
dawn_nautical	航海晨光开始，已有足够光线辨别地平线和部分物体；太阳在地平线下 12 度。
dawn_civil	民用晨光开始，光线足以辨认物体并开始户外活动；太阳在地平线下 6 度。
sunrise	早晨太阳上缘出现在东方地平线上。
solar_noon	太阳当天位于地平线上最高的位置。
sunset	傍晚太阳的最后边缘消失在西方地平线下。
dusk_civil	民用暮光结束，仍能辨认物体，部分恒星和行星可见；太阳在地平线下 6 度。
dusk_nautical	太阳在地平线下 12 度，物体无法辨认，肉眼看不到地平线。
dusk_astronomical	天文暮光结束，天空完全黑暗；太阳在地平线下 18 度。

所有太阳事件都用 UTC 计算，不受配置时区影响。

极地地区并非每天都有日出或日落，调度器能够处理：没有日出的那一天，不运行日出任务。唯一例外是 solar_noon，其严格定义是太阳通过天球子午圈的时刻，因此即使太阳始终在地平线下，也每天发生。

曙暮光指黎明至日出、日落至黄昏之间的时段。根据民用、航海或天文定义，以及希望在时段开始还是结束执行，可从上表选择对应事件。

更多说明见 celery.schedules.solar。

启动调度器

启动 beat 服务：

$ celery -A proj beat

也可以使用 worker 的 -B 选项将 beat 嵌入工作进程。确定永远只运行一个工作节点时很方便，但这种方式不常使用，因此不推荐用于生产：

$ celery -A proj worker -B

Beat 需要将任务上次运行时间写入本地数据库文件，默认名为 celerybeat-schedule，因此必须具有当前目录写权限。也可以指定其他位置：

$ celery -A proj beat -s /home/celery/var/run/celerybeat-schedule

以守护进程运行的方法见 守护进程配置。

使用自定义调度器类

通过命令行 --scheduler 指定自定义调度器。

默认是 celery.beat.PersistentScheduler，用本地 shelve 数据库记录最近运行时间。

django-celery-beat 扩展将计划保存在 Django 数据库，并提供便捷的管理界面，可在运行期间管理周期任务。

安装和使用步骤：

1. 使用 pip 安装：

$ pip install django-celery-beat

2. 在 Django 项目 settings.py 的 INSTALLED_APPS 中添加 django_celery_beat：

INSTALLED_APPS = (
    ...,
    'django_celery_beat',
)

注意，模块名只有下划线，没有短横线。

3. 应用数据库迁移，创建所需表：

$ python manage.py migrate

4. 使用数据库调度器启动 beat：

$ celery -A proj beat -l INFO --scheduler django_celery_beat.schedulers:DatabaseScheduler

也可以直接将该值写入 beat_scheduler 设置。

5. 打开 Django 管理界面，配置周期任务。

---

原文：Periodic Tasks。作者／维护者：Celery 文档贡献者。本文为原文的中文译文；代码保留原文内容。