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

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()任务里面虽然。此外，这个例子是过分简单化的，我确实需要在所有页面都被输入后做一些事情（例如在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)