celery - 将生成的列表映射到 celery 中的任务的最佳方法

Question

我正在寻找一些关于将任务生成的列表映射到 celery 中的另一个任务的最佳方法的建议。

假设我有一个名为 parse 的任务，它解析 PDF 文档并输出页面列表。然后，每个页面都需要单独传递给另一个名为 feed 的任务。 .这一切都需要进入一个名为 process 的任务中。

所以，我可以这样做的一种方法是:

@celery.task
def process:
    pages = parse.s(path_to_pdf).get()

    feed.map(pages)

当然，这不是一个好主意，因为我调用 get()在一个任务里面。

此外，这是低效的，因为我的 parse任务被包裹在一个生成器函数中并且能够产生页面，这意味着它应该可以在解析器产生最后一页之前将第一页排队以供馈送。

另一种可能性是这样做:

@celery.task
def process:
    for page in parse.s(path_to_pdf).get():
        feed.delay(page)

该示例仍然涉及调用 get()虽然在一个任务里面。此外，这个例子过于简单化了，我真的需要在所有页面都被输入之后做一些事情(即在 chord 中)。

我正在寻找在 celery 中做到这一点的最佳方法。我会很感激任何建议。

谢谢!

Answer 1

这对你来说可能已经太晚了，但你可能想要使用任务链:

@celery.task
def process():
    return chain(parse.s(), feed_map.s())

@celery.task
def feed_map(pages):
    return feed.map(pages)

如果你有一些最后的任务，说 final ，你可以这样做:

@celery.task
def feed_map(pages):
    return chord(feed.map.s(page) for page in pages, final.s)