Периодические задачи

celery beat — это планировщик. Он запускает задачи с регулярными интервалами, после чего их выполняют доступные рабочие узлы кластера.

По умолчанию записи берутся из настройки beat_schedule, но можно использовать и пользовательские хранилища, например хранить записи в базе данных SQL.

Необходимо убедиться, что для одного расписания в каждый момент времени работает только один планировщик, иначе задачи будут дублироваться. При централизованном подходе синхронизировать расписание не требуется, и служба может работать без блокировок.

По умолчанию для расписаний периодических задач используется часовой пояс UTC, но его можно изменить с помощью настройки timezone.

Например, можно указать часовой пояс Europe/London:

timezone = 'Europe/London'

Эту настройку необходимо добавить в приложение: либо задать напрямую с помощью (app.conf.timezone = 'Europe/London'), либо добавить в модуль конфигурации, если вы создали его с помощью app.config_from_object. Дополнительную информацию о параметрах конфигурации см. в разделе Конфигурация.

Планировщик по умолчанию (сохраняющий расписание в файле celerybeat-schedule) автоматически обнаружит изменение часового пояса и сбросит расписание. Другие планировщики могут быть не такими интеллектуальными (например, планировщик базы данных Django, см. ниже), и в этом случае расписание придётся сбросить вручную.

$ pythonmanage.pyshell
>>> from djcelery.models import PeriodicTask
>>> PeriodicTask.objects.update(last_run_at=None)

$ pythonmanage.pyshell
>>> 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(). Обратите внимание: сигнал on_after_configure отправляется после настройки приложения, поэтому задачи вне модуля, в котором объявлено приложение (например, в файле tasks.py, обнаруживаемом с помощью celery.Celery.autodiscover_tasks()), должны использовать более поздний сигнал, например on_after_finalize.

Функция add_periodic_task() добавит запись в настройку beat_schedule в фоновом режиме. Эту же настройку можно использовать для добавления периодических задач вручную:

Пример: запуск задачи tasks.add каждые 30 секунд.

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

Использование timedelta в качестве расписания означает, что задача будет отправляться с интервалом в 30 секунд (первая задача будет отправлена через 30 секунд после запуска celery beat, а затем — каждые 30 секунд после предыдущего запуска).

Также можно использовать расписание типа Crontab; см. раздел Расписания Crontab.

Как и в случае с cron, задачи могут выполняться одновременно, если первая задача не завершится до запуска следующей. Если это вызывает опасения, используйте стратегию блокировки, чтобы гарантировать одновременный запуск только одного экземпляра (например, см. раздел Как гарантировать, что задача выполняется только один раз одновременно).

Если вам нужно точнее задавать время выполнения задачи, например определённое время суток или день недели, можно использовать тип расписания 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 очень гибок.

Примеры:

Дополнительную документацию см. в разделе 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)

Убедитесь, что для широты и долготы указан правильный знак:

Возможные типы событий:

Все солнечные события рассчитываются по UTC, поэтому на них не влияет настройка часового пояса.

В полярных регионах солнце может восходить или заходить не каждый день. Планировщик умеет обрабатывать такие случаи (например, событие sunrise не будет выполнено в день, когда солнце не восходит). Исключение составляет solar_noon — это событие формально определяется как момент прохождения солнца через небесный меридиан и происходит каждый день, даже если солнце находится ниже горизонта.

Сумерки определяются как период между рассветом и восходом солнца, а также между закатом и наступлением темноты. Можно запланировать событие на «сумерки» в соответствии с вашим определением сумерек (гражданские, навигационные или астрономические), а также выбрать, должно ли событие происходить в начале или в конце сумерек, указав соответствующее событие из списка выше.

Дополнительную документацию см. в разделе celery.schedules.solar.

Чтобы запустить службу celery beat:

$ celery-Aprojbeat

Также можно встроить beat в рабочий процесс, включив параметр -B. Это удобно, если вы не планируете запускать более одного рабочего узла, но такой способ используется нечасто и поэтому не рекомендуется для эксплуатации в производственной среде:

$ celery-Aprojworker-B

Beat должен хранить время последнего запуска задач в локальном файле базы данных (по умолчанию он называется celerybeat-schedule), поэтому ему нужен доступ на запись в текущий каталог. Также можно указать для этого файла другое расположение:

$ celery-Aprojbeat-s/home/celery/var/run/celerybeat-schedule