在所有其他任务完成后需要一个芹菜任务开始[重复]

Question

我对 Celery 比较陌生。我有一个要求，只有在所有其他任务完成后才开始我的芹菜任务之一。我确实玩过它，并且还在网上找到了一些文档，其中表明我应该使用 groups 和 chords 等，我无法完全理解。

我还发布了另一个关于 SO here 的问题，但到目前为止还没有找到任何令人信服的答案。

最后我遇到了this SO question，我可以清楚地（至少在概念上）理解接受的答案中发生了什么。但是，当我尝试在最后复制完全相同的程序时，它抛出了以下错误：

EncodeError: <AsyncResult: cf5875f1-7f72-449c-9808-07c9c9459737> is not JSON serializable

我在 cmets 以及上述问题中也发布了相同的内容，这个问题似乎很老了，因此到目前为止我的评论还没有收到任何关注。

有人可以帮忙吗？

Answer 1

this SO answer 中接受的答案实际上帮助我实现了我想要的。 （可能这是一种 hacky 方式，但它解决了我的问题）。然而，答案抛出了一个错误：

EncodeError: <AsyncResult: cf5875f1-7f72-449c-9808-07c9c9459737> is not JSON serializable

我能够解决的问题如下（我在那个答案的 cmets 中也提到了这个变化）

所以在答案的用例部分：

tasks = []
for i in xrange(10):
    tasks.append(power.delay(i, 2))

amass.delay([], tasks)

作为amass.delay() 的第二个参数，我们传递了一个任务对象列表。我只是将其更改为实际传递任务 ID 列表。所以amass() 现在看起来像：

tasks = []
for i in xrange(10):
    x = power.delay(i, 2)
    tasks.append(x.id)

amass.delay([], tasks)

并在amass()中进行了相应的更改，如下所示 它解决了错误并完成了我试图实现的工作。

@celery.task()
def amass(results, tasks):
    completed_tasks = []
    for task_id in tasks:
        result = AsyncResult(task_id, app=celery)
        if result.ready():
            completed_tasks.append(task_id)
            #results.append(task.get()) did not need this so commented it out

    # remove completed tasks
    tasks = list(set(tasks) - set(completed_tasks))

    if len(tasks) > 0:
        # resend the task to execute at least 1 second from now
        amass.delay(results, tasks, countdown=1)
    else:
        # we done
        print results

Answer 2

如果

您已经有一个必须一个接一个地执行的任务列表，并且想要合并您必须使用的结果，仅chords。

否则

如果您不断向 celery 添加任务（不是立即），并且希望它们一个一个地执行，那么只需设置 1 个具有 concurrency=1 的 worker 就可以实现这一点，这些任务只能以串行方式执行。

我之前也遇到过并行执行的问题，一开始是用concurrency=1运行一个worker解决的，后来我用redis锁机制解决了。