python - 回调 celery apply_async

Question

我在我的应用程序中使用 celery 来运行周期性任务。让我们看下面的简单示例

from myqueue import Queue
@perodic_task(run_every=timedelta(minutes=1))
def process_queue():
    queue = Queue()
    uid, questions = queue.pop()
    if uid is None:
        return

    job = group(do_stuff(q) for q in questions)
    job.apply_async()

def do_stuff(question):
    try:
        ...
    except:
        ...
        raise

正如您在上面的示例中看到的，我使用 celery 来运行异步任务，但是(因为它是一个队列)我需要执行 queue.fail(uid) 如果在 do_stuff 或 queue.ack(uid) 中出现异常，否则。在这种情况下，在这两种情况下从我的任务中获得一些回调是非常清楚和有用的 - on_failure 和 on_success。

我看到了一些documentation ，但从未见过将回调与 apply_async 结合使用的做法。有可能吗？

Answer 1

子类化 Task 类并重载 on_success 和 on_failure 函数:

from celery import Task


class CallbackTask(Task):
    def on_success(self, retval, task_id, args, kwargs):
        '''
        retval – The return value of the task.
        task_id – Unique id of the executed task.
        args – Original arguments for the executed task.
        kwargs – Original keyword arguments for the executed task.
        '''
        pass
        
    def on_failure(self, exc, task_id, args, kwargs, einfo):
        '''
        exc – The exception raised by the task.
        task_id – Unique id of the failed task.
        args – Original arguments for the task that failed.
        kwargs – Original keyword arguments for the task that failed.
        '''
        pass

使用:

@celery.task(base=CallbackTask)  # this does the trick
def add(x, y):
    return x + y