Има опашки за грешки в celery

Е, не можете да използвате механизма retry, ако искате да насочите задачата към друга опашка. От документите:

retry() може да се използва за повторно изпълнение на задачата, например в случай на възстановими грешки.

Когато извикате повторен опит, той ще изпрати ново съобщение, използвайки същия идентификатор на задача, и ще се погрижи да се увери, че съобщението е доставено на същата опашка като първоначалната задача.

Ще трябва да рестартирате себе си и да го насочите ръчно към желаната от вас опашка в случай на повдигнато изключение. Изглежда добра работа за обратни извиквания за грешки.

Основният проблем е, че трябва да получим името на задачата в обратното извикване на грешката, за да можем да я стартираме. Също така може да не искаме да добавяме обратното извикване всеки път, когато стартираме задача. По този начин декораторът би бил добър начин за автоматично добавяне на правилното обратно извикване.

from functools import partial, wraps

import celery


@celery.shared_task
def error_callback(task_id, task_name, retry_queue, retry_routing_key):
    # We must retrieve the task object itself.
    # `tasks` is a dict of 'task_name': celery_task_object
    task = celery.current_app.tasks[task_name]
    # Re launch the task in specified queue.
    task.apply_async(queue=retry_queue, routing_key=retry_routing_key)


def retrying_task(retry_queue, retry_routing_key):
    """Decorates function to automatically add error callbacks."""
    def retrying_decorator(func):
        @celery.shared_task
        @wraps(func)  # just to keep the original task name
        def wrapper(*args, **kwargs):
            return func(*args, **kwargs)
        # Monkey patch the apply_async method to add the callback.
        wrapper.apply_async = partial(
            wrapper.apply_async,
            link_error=error_callback.s(wrapper.name, retry_queue, retry_routing_key)
        )
        return wrapper
    return retrying_decorator


# Usage:
@retrying_task(retry_queue='another_queue', retry_routing_key='another_routing_key')
def failing_task():
    print 'Hi, I will fail!'
    raise Exception("I'm failing!")

failing_task.apply_async()

Можете да настроите декоратора, за да предава каквито параметри имате нужда.

Имах подобен проблем и го реших може да не е по най-ефективния начин, но въпреки това моето решение е следното:

Създадох django модел, за да запазя всички мои идентификационни номера на задачите от целина и който може да проверява състоянието на задачата.

След това създадох друга задача за целина, която се изпълнява в безкраен цикъл и проверява всички задачи, които се изпълняват, за действителното им състояние и ако състоянието е „НЕУСПЕШНО“, просто го изпълнява отново. Всъщност не променям опашката за задачата, която стартирам отново, но мисля, че можете да приложите някаква персонализирана логика, за да решите къде да поставите всяка задача, която стартирате отново по този начин.