使用依赖图执行 Celery 任务答案

【问题标题】：Executing Celery tasks with dependency graph使用依赖图执行 Celery 任务
【发布时间】：2011-09-28 16:01:58
【问题描述】：

我希望 Celery 任务取决于 2 个或更多其他任务的结果。我研究了 Python+Celery: Chaining jobs? 和 http://pypi.python.org/pypi/celery-tasktree ，但只有当任务只有一个依赖任务时，这些才是好的。

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

有什么建议吗？

【问题讨论】：

标签： python celery

【解决方案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

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

【讨论】：

【解决方案2】：

mrbox 为真，您可以重试直到结果准备好，但在文档中并不清楚，当您重试时，您必须传递 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")

【讨论】：

【解决方案3】：

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

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

【讨论】：