如何使用 celery AsyncTask 的结果回调

Question

我有以下情况。

我client向工作人员发送以下任务：

# client
task = my_task.apply_async((some_params), queue='my_queue')
# task.get()  # This blocks

我的工作人员正确执行任务并返回结果。

因此，检索结果可以task.get()工作，但会阻塞。现在我想要的是当结果（成功或失败）可用时调用的回调。

Task类有一个on_success函数。但这是用在工人身上的。类似问题

有什么想法或解决方案吗？

Answer 1

您可以对任务进行回调，但不能通知或回调调用者或客户端（因为 celery 不在进程中），而是必须将另一个 celery 任务用作回调。 如果您想使用回调，可以使用 celery 的链接或拉票功能。

对于使用启动任务结果的简单回调，您可以执行以下操作：

@app.task
def add(m, n):
    return m + n

@app.task
def callback(result):
    print(f'My result was {result}')

def client_caller():
    add.apply_async(args=(2, 2), link=callback.s())