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。

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

Answer 1

子类Task类并重载on_success和on_failure函数：

class CallbackTask(Task):
    def on_success(self, retval, task_id, args, kwargs):
        pass

    def on_failure(self, exc, task_id, args, kwargs, einfo):
        pass


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