将n个任务添加到芹菜队列并等待结果

Question

我会将多个任务添加到芹菜队列并等待结果.我有各种想法如何利用某种形式的共享存储(memcached,redis,db等)实现这一点,但是,我会认为这是Celery可以自动处理但我无法在线找到任何资源的东西.

代码示例

def do_tasks(b):
    for a in b:
        c.delay(a)

    return c.all_results_some_how()

Answer 1

对于Celery> = 3.0,不推荐使用 TaskSet 以支持group.

from celery import group
from tasks import add

job = group([
             add.s(2, 2),
             add.s(4, 4),
             add.s(8, 8),
             add.s(16, 16),
             add.s(32, 32),
])

在后台启动该组:

result = job.apply_async()

等待:

result.join()

Answer 2

Task.delay回报AsyncResult.使用AsyncResult.get得到每个任务的结果.

为此,您需要保留对任务的引用.

def do_tasks(b):
    tasks = []
    for a in b:
        tasks.append(c.delay(a))
    return [t.get() for t in tasks]

或者您可以使用ResultSet:

def do_tasks(b):
    rs = ResultSet([])
    for a in b:
        rs.add(c.delay(a))
    return rs.get()