从 celery 任务中调用 async_result.get()

Question

我有一个 celery 任务调用另一个 remote 任务（它在另一个 celery 应用程序上，在另一个服务器上......）。 当我尝试从我的任务中 .get() 远程任务的结果时，如下所示：

@app.task()
def my_local_task():
  result_from_remote = app.send_task('remote_task', [arg1, arg2])
  return result_from_remote.get()

我收到此错误：

RuntimeWarning: Never call result.get() within a task! See http://docs.celeryq.org/en/latest/userguide/tasks.html#task-synchronous-subtasks

In Celery 3.2 this will result in an exception being
raised instead of just being a warning.

  warnings.warn(RuntimeWarning(E_WOULDBLOCK))

基本上我希望我的任务是“同步的”——我希望它等待远程任务的结果，我可以接受。

我可以告诉 celery 可以吗？ 当然有链接的解决方案，除了不可能链接远程任务。调用远程任务的唯一方法是使用 app.send_task，它返回一个 AsyncResult，并且我无法链接，因为我需要任务函数本身..

Answer 1

如果你希望你的任务是同步的，你可以使用ready()来保持一个循环：

import time

while not result_from_remote.ready():
    time.sleep(5)

return result_from_remote.get()

Answer 2

当然有链接的解决方案，除了不可能链接远程任务。调用远程任务的唯一方法是使用 app.send_task，它返回一个 AsyncResult，并且我无法链接，因为我需要任务函数本身..

不，可以链接远程任务。我刚刚在我的一个项目中尝试过，它可以工作。我建议你先用一个简单的测试任务来尝试它，以确保在转向更复杂的东西之前你已经掌握了基础知识。我创建了这些任务：

@app.task
def foo(arg):
    return arg + 1

@app.task
def bar(arg):
    return "I barred " + str(arg)

这两个任务保存在一个名为app.tasks 的模块中。 （它是 Django 项目的一部分。）

然后我写了一个命令：

import celery
print (celery.signature("app.tasks.foo", args=(1, )) |
       celery.signature("app.tasks.bar")).delay().get()

我登上了屏幕：

I barred 2

Answer 3

如果您知道自己正在做的事情是安全的，这里有一个片段可以消除警告：

from celery.result import allow_join_result

with allow_join_result():
    result.get()

source

Answer 4

这适用于 celery 4

@app.task()
def my_local_task():
    result_from_remote = app.send_task('remote_task', [arg1, arg2])

    while not result_from_remote.ready():
        time.sleep(1)

    return result_from_remote.result