芹菜发子任务

Question

我的“大”任务可以分步骤工作:它可以终止任务，也可以生成更多任务。在我的例子中，我数到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. 当任务异步执行时，如何获得中间状态(在本例中是介于1和5之间的数字)？

Answer 1

以我为例，

@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. 在芹菜监视器里，我可以看到收到的任务。 接收任务: services.tasker.test67f22e02-7c39-4c1d-a646-acabeb72d208 1接收任务: services.tasker.test9eed6d45-4931-4790-8477-3bbe75e213e4 任务services.tasker.test67f22e02-7c39-4c1d-a646-acabeb72d208在3.021544273 s中成功:无 2 ..。

1. 可以使用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