使用依赖图执行 Celery 任务

Question

我希望 Celery 任务依赖于 2 个或更多其他任务的结果。我研究过 Python+Celery：链接作业？ 和 http://pypi.python.org/pypi/celery-tasktree ，但只有当任务只有一个依赖任务时，这些才是好的。

我了解 TaskSet，但似乎没有办法在 TaskSetResult.ready() 变为 True 时立即执行回调。我现在想到的是有一个定期任务，每隔几毫秒左右轮询一次 TaskSetResult.ready() 并在返回 True 时触发回调，但这对我来说听起来相当不优雅。

有什么建议吗？

Answer 1

在最新版本的 Celery (3.0+) 中，您可以使用所谓的和弦来实现所需的效果：

来自 http://docs.celeryproject.org/en/latest/userguide/canvas.html#the-primitives：

简单和弦

和弦原语使我们能够添加在所有情况下调用的回调
组中的任务已完成执行，这通常是
对于不太并行的算法来说是必需的：

 >>> from celery import chord
 >>> res = chord((add.s(i, i) for i in xrange(10)), xsum.s())()
 >>> res.get()
 90

免责声明：我自己还没有尝试过这个。

Answer 2

mrbox 是 true，您可以重试，直到结果准备好，但文档中不太清楚，当您重试时，您必须传递 setid 和子任务元素，并且为了恢复它，您必须使用下面的 map 函数是一个示例代码，用于解释我的意思。

def run(self, setid=None, subtasks=None, **kwargs):

    if not setid or not subtasks:
        #Is the first time that I launch this task, I'm going to launch the subtasks
        …
        tasks = []
        for slice in slices:
            tasks.append(uploadTrackSlice.subtask((slice,folder_name)))

        job = TaskSet(tasks=tasks)
        task_set_result = job.apply_async()
        setid = task_set_result.taskset_id
        subtasks = [result.task_id for result in task_set_result.subtasks]
        self.retry(exc=Exception("Result not ready"), args=[setid,subtasks])

    #Is a retry than we just have to check the results        
    tasks_result = TaskSetResult(setid, map(AsyncResult,subtasks))
    if not tasks_result.ready():
        self.retry(exc=Exception("Result not ready"), args=[setid,subtasks])
    else:    
        if tasks_result.successful():
            return tasks_result.join()
        else:
            raise Exception("Some of the tasks was failing")

Answer 3

恕我直言，你可以做类似于文档中所做的事情 -

您可以使用 max_retries=None 的重试方法 - 如果“基本”任务之一 .ready() 为 false，您可以触发 .retry() 方法，直到两项“基本”任务均已完成。