Разрешить выполнение задачи, если она еще не запланирована с использованием сельдерея

Я использую Celery для управления планированием задач в приложении Django, которое я разрабатываю, я работаю с базой данных Django только для тестирования.

Я просто попробовал несколько действий для обработки выполнения задачи, только если она еще не запланирована или не выполнена, как предлагаемая в этой статье , но пока ничего не сделано.

Что-то вроде этого:

task.py

@task()
def add(x, y):
   return x + y

И затем, когда вы вызываете его дважды, как следующим образом:

import myapp.tasks.add

myapp.tasks.add.apply_async((2,2), task_id=1, countdown=15)
myapp.tasks.add.apply_async((2,2), task_id=2, countdown=15)

Должно быть разрешено использование одного экземпляра в countdown=15. Как я могу добиться, чтобы второй вызов никогда не выполнял его, если есть другой запуск или ожидание?

Ответы

Ответ 1

Одна из проблем с принятым ответом заключается в том, что он медленный. Проверка того, что задача уже запущена, включает вызов брокера, а затем повторение как текущих, так и активных задач. Если вы хотите быстро поставить задачу в очередь, это не сработает. Также текущее решение имеет небольшое условие гонки, в котором 2 процесса могут проверять, была ли задача поставлена ​​в очередь одинаково (выясните, что это не так), которая затем поставит в очередь 2 задачи.

Лучшим решением будет то, что я называю дебютными задачами. В основном вы увеличиваете счетчик каждый раз, когда вы ставите в очередь задачу. Когда задача начинается, вы уменьшаете ее. Используйте redis, а затем все атомы.

например.

Задайте задачу:

conn = get_redis()
conn.incr(key)
task.apply_async(args=args, kwargs=kwargs, countdown=countdown)

Затем в задаче у вас есть 2 варианта, вы хотите выполнить задачу через 15 секунд после того, как первая была поставлена ​​в очередь (дроссель) или выполнить ее через 15 секунд после того, как последняя была поставлена ​​в очередь (debounce). То есть, если мы продолжаем пытаться выполнить ту же задачу, мы расширяем таймер или просто ждем 15 для первого и игнорируем остальные задачи, которые были поставлены в очередь.

Легко поддерживать оба, вот отброс, где мы ждем, пока задачи перестанут попадать в очередь:

conn = get_redis()
counter = conn.decr(key)
if counter > 0:
    # task is queued
    return
# continue on to rest of task

Версия дроссельной заслонки:

counter = conn.getset(key, '0')
if counter == '0':
    # we already ran so ignore all the tasks that were queued since
    return
# continue on to task

Другим преимуществом этого решения над принятым является то, что ключ полностью находится под вашим контролем. Поэтому, если вы хотите, чтобы одна и та же задача выполнялась, но только один раз для разных id/объектов, например, вы включаете это в свой ключ.

Update

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

Throttle v2 (при постановке задачи в очередь)

conn = get_redis()
counter = conn.incr(key)
if counter == 1:
    # queue up the task only the first time
    task.apply_async(args=args, kwargs=kwargs, countdown=countdown)

Затем в задаче вы установите счетчик обратно на 0.

Вам даже не нужно использовать счетчик, если у вас есть набор, вы можете добавить ключ к набору. Если вы вернетесь на 1, тогда ключ не был установлен, и вы должны поставить в очередь задачу. Если вы вернетесь 0, то ключ уже находится в наборе, поэтому не ставьте в очередь задачу.

Ответ 2

Посмотрите, прежде чем прыгать! Вы можете проверить, есть ли какие-либо задачи, выполняемые/ожидающие, перед очередью задач.

from celery.task.control import inspect

def is_running_waiting(task_name):
    """
    Check if a task is running or waiting.
    """
    scheduled_tasks = inspect().scheduled().values()[0]
    for task in scheduled_tasks:
        if task['request']['name'] == task_name:
            return True
    running_tasks = inspect().active().values()[0]
    for task in running_tasks:
        if task['request']['name'] == task_name:
            return True

Теперь, если вы ставите в очередь три задачи добавления, первый из них будет поставлен в очередь для выполнения, остальные не будут поставлены в очередь.

for i in range(3):
    if not is_running_waiting('add'):
        add.apply_async((2,2), countdown=15)