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

Question

我正在寻找一些建议,以便将从任务生成的列表映射到芹菜中的另一个任务的最佳方法.

假设我有一个名为的任务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()任务内部.此外,这个例子过于简单化,我真的需要在所有页面都被输入之后做一些事情(即在a中chord).

我正在寻找在芹菜中做到这一点的最佳方式.我很感激任何建议.

谢谢!

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)