使用 Celery 作为管道答案

【问题标题】：Use Celery for a pipeline使用 Celery 作为管道
【发布时间】：2014-07-01 06:08:07
【问题描述】：

这是将 Celery 用于管道的最佳方式：TaskA -> TaskB -> TaskC？我对 Celery 一无所知，文档并没有过多解释在实际应用中应该如何做。

@app.task
def taskA(t_id):
    if success:
        TaskB.delay(t_id)
    else:
        ReportError.delay(t_id)

@app.task
def taskAA(t_id):
    if success:
        TaskB.delay(t_id)
    else:
        ReportError.delay(t_id)

@app.task
def taskB(t_id):
    if success:
        TaskC.delay(t_id)
    else:
        ReportError.delay(t_id)

也许我不应该将 Celery 用于此类任务...

【问题讨论】：

stackoverflow.com/questions/3901101/pythoncelery-chaining-jobs的可能重复
使用芹菜链 celery.readthedocs.org/en/latest/userguide/canvas.html#chains 已在上一条评论的链接答案中描述。
@MauroRocco 我不喜欢 Chain 的一件事是它是连续的，没有任何分支。例如，对于taskA，我可能需要启动一个图像下载器和一个视频下载器，在视频下载器之后我需要一个分段器。也许 celery 不是这种任务的正确框架？我正在寻找一种静态构建系统流程的方法，我可以将 t_id 提供给 taskA，一切都会正常工作。
@YifanZhang 我认为您应该认真阅读此页面celery.readthedocs.org/en/latest/userguide/canvas.html。 Celery 允许您构建不同的工作流类型，例如，您可以使用 group 和 chord 一起进行一些并行工作，然后调用聚合的最终任务。

标签： python celery

【解决方案1】：

正如其他评论者所说，您可以使用链。有关更多信息，请参阅https://celery.readthedocs.org/en/latest/userguide/canvas.html#chains。根据您的原始问题，如下所示：

from celery import chain

ret = chain(taskA.s(), taskB.s(), taskC.s()).apply_async()

if ret.status == u'SUCCESS':
    print "result:", ret.get()

【讨论】：