芹菜从子任务发行任务



我的"大"任务在步骤中起作用:它可以终止或产生更多任务。在我的示例中,我数量为5。

from celery import Celery
app = Celery('tasks', broker='redis://localhost:6379/0')
from time import sleep
@app.task
def slow_add(x):
    """Slowly counts to 5"""
    sleep(1)
    print (x)
    if x == 5:
        return x
    else:
        return slow_add.s(x+1)()

当我安排任务时,我只会得到一个调用:

In [48]: asks.slow_add.run(1)
1
2
3
4
5
Out[48]: 5
  1. 如何将其异步称为?我尝试了apply_async的不同变体,延迟,但无济于事。
  2. 我在芹菜监视器中没有看到任何任务。为什么?
  3. 在任务是异步执行时,我如何获得中间状态(在这种情况下,数字之间的数字)?

在我的示例中,

@app.task
def test(x):
    import time
    time.sleep(1)
    print x
    if x == 5:
        return x
    else:
        return test.delay(x+1)
  1. 当我使用res = test.delay(1)
  2. 时,它的工作正常很好
  3. 在芹菜监视器中,我可以看到接收到的任务

    接收任务:services.tasker.test [67F22E02-7C39-4C1D-A646-ACABEB72D208]
    1
    收到的任务:services.tasker.test [9EED6D45-4931-4790-8477-3BBE75E213E4]
    task Services.Tasker.test [67F22E02-7C39-4C1D-A646-ACABEB72D208]在3.021544273S中取得了成功

  4. 您可以使用res.ready()获得任务状态,当任务完成时返回True,否则返回False

在我的测试案例中,结果就像

>>> res = test.delay(1)
>>> res.ready()  # before finishing task
False
>>> res.ready()  # after finishing task
True

def final_result(r):
    if isinstance(r.result, celery.result.AsyncResult):
         return final_result(r.result)
    else:
        return r.result

并在上面使用

>>> print final_result(res)
5

相关内容

  • 没有找到相关文章

最新更新