График работы сельдерея (Celery, Django и RabbitMQ)

Я хочу иметь задачу, которая будет выполняться каждые 5 минут, но она будет ждать завершения последнего выполнения, а затем начнет считать это 5 минут. (Таким образом, я также могу быть уверен, что работает только одна задача). Самый простой способ, который я нашел, - запустить django application manage.py shell и запустить это:

while True:
    result = task.delay()
    result.wait()
    sleep(5)

но для каждой задачи, которую я хочу выполнить таким образом, я должен запустить ее собственную оболочку, есть ли простой способ сделать это? Может быть, какой-то королевский заказчик из django celery scheduler?

Ответ 1

Все, что вам нужно, это указать в задаче celery conf witch, которую вы хотите запускать периодически и с каким интервалом.

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

from datetime import timedelta

CELERYBEAT_SCHEDULE = {
    "runs-every-30-seconds": {
        "task": "tasks.add",
        "schedule": timedelta(seconds=30),
        "args": (16, 16)
     },
}

Помните, что вам нужно запускать сельдерей в режиме бит с опцией -B

manage celeryd -B

Вы также можете использовать стиль crontab вместо временного интервала, проверьте это:

http://ask.github.com/celery/userguide/periodic-tasks.html

Если вы используете django-сельдерей, помните, что вы можете также использовать django db в качестве планировщика для периодических задач, таким образом вы можете легко добавить через новые периодические задачи панели администратора django-celery. Для этого вам нужно установить планировщик celerybeat в settings.py таким образом

CELERYBEAT_SCHEDULER = "djcelery.schedulers.DatabaseScheduler"

Ответ 2

Удивительно, как никто не понимает этот вопрос человека. Они периодически задают вопрос о запуске задач, но как обеспечить, чтобы сельдерей не запускал одновременно два экземпляра одной и той же задачи. Я не думаю, что есть способ сделать это с Celery напрямую, но то, что вы можете сделать, - это то, что одна из задач получает право блокировки сразу же, когда она начинается, и если это не удается, попробуйте снова через несколько секунд (используя повторную попытку), Задача освободит блокировку до ее возвращения; вы можете сделать автоматическое выключение блокировки через несколько минут, если он когда-либо выйдет из строя или истечет время ожидания.

Для блокировки вы, вероятно, можете просто использовать свою базу данных или что-то вроде Redis.

Ответ 3

Вам может быть интересен этот более простой метод, который не требует изменений в celery conf.

@celery.decorators.periodic_task(run_every=datetime.timedelta(minutes=5))
def my_task():
    # Insert fun-stuff here

Ответ 4

Чтобы расширить сообщение @MauroRocco, от http://docs.celeryproject.org/en/v2.2.4/userguide/periodic-tasks.html

Использование timedelta для расписания означает, что задача будет выполнена через 30 секунд после запуска celerybeat, а затем каждые 30 секунд после последнего прогона. Также существует график, подобный crontab, см. Раздел о расписаниях Crontab.

Таким образом, это действительно достигнет цели, которую вы хотите.

Ответ 5

Из-за того, что celery.decorators устарели, вы можете использовать decorator_task:

from celery.task.base import periodic_task
from django.utils.timezone import timedelta

@periodic_task(run_every=timedelta(seconds=5))
def my_background_process():
    # insert code

Ответ 6

Добавьте эту задачу в отдельную очередь, а затем используйте отдельный рабочий для этой очереди с параметром concurrency, установленным в 1.