Как узнать, была ли задача уже поставлена ​​в очередь в django-сельдерее?

Question

Вот мои настройки:

• Джанго 1,3
• сельдерей 2.2.6
• Джанго-сельдерей 2.2.4
• djkombu 0.9.2

В моем файле settings.py у меня есть

BROKER_BACKEND = "djkombu.transport.DatabaseTransport"

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

Теперь к моей проблеме: у меня есть задание, инициированное пользователем, которое может занять несколько минут. Я хочу, чтобы задача выполнялась только один раз для каждого пользователя, и я буду кэшировать результаты задачи во временном файле, поэтому, если пользователь снова запускает задачу, я просто возвращаю кэшированный файл. У меня есть код, который выглядит следующим образом в моей функции просмотра:

task_id = "long-task-%d" % user_id
result = tasks.some_long_task.AsyncResult(task_id)

if result.state == celery.states.PENDING:
    # The next line makes a duplicate task if the user rapidly refreshes the page
    tasks.some_long_task.apply_async(task_id=task_id)
    return HttpResponse("Task started...")
elif result.state == celery.states.STARTED:
    return HttpResponse("Task is still running, please wait...")
elif result.state == celery.states.SUCCESS:
    if cached_file_still_exists():
        return get_cached_file()
    else:
        result.forget()
        tasks.some_long_task.apply_async(task_id=task_id)
        return HttpResponse("Task started...")

Этот код почти работает. Но я сталкиваюсь с проблемой, когда пользователь быстро перезагружает страницу. Существует задержка в 1-3 секунды между тем, когда задача ставится в очередь, и когда задача, наконец, снимается с очереди и передается работнику. В течение этого времени состояние задачи остается в состоянии ОЖИДАНИЯ, в результате чего логика представления запускает дублирующую задачу.

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

Answer 1

Я решил это с Redis. Просто установите ключ в redis для каждой задачи, а затем удалите ключ из redis в методе after_return задачи. Redis легкий и быстрый.

Answer 2

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

http://docs.celeryproject.org/en/latest/tutorials/task-cookbook.html#cookbook-task-serial

Django может быть настроен на использование memcached в качестве бэкэнда кеша, и его можно распределить по нескольким машинам ... мне кажется, это лучше. Мысли?

Answer 3

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

Например, если используется СУБД (таблица со столбцами - task_id, state, result):

Просмотр части:

1. Использовать управление транзакциями.
2. Используйте SELECT FOR UPDATE, чтобы получить строку, где task_id == "long-task-% d"% user_id. SELECT FOR UPDATE будет блокировать другие запросы до тех пор, пока этот запрос не завершится или не будет выполнен ROLLBACK.
3. Если его не существует - установите состояние PENDING и запустите some_long_task, завершите запрос.
4. Если состояние В ОЖИДАНИИ - сообщите пользователю.
5. Если состояние УСПЕХ - установите состояние в ОЖИДАНИЕ, запустите задачу, верните файл, на который указывает столбец «результат». Я основываю это на предположении, что вы хотите перезапустить задачу, чтобы получить результат. COMMIT
6. Если состояние - ОШИБКА - установите состояние в ОЖИДАНИЕ, запустите задание, сообщите пользователю. COMMIT

Задание:

1. Подготовьте файл, заверните в try, перехватите блок.
2. В случае успеха - ОБНОВИТЬ соответствующую строку с состоянием = УСПЕХ, результат.
3. При сбое - ОБНОВИТЬ соответствующую строку с состоянием = ОШИБКА.