调用celery.task.chunks的正确方法是什么？

Question

我正在用Python3和芹菜做一个并行的任务。我喜欢把它分成几块，这样就可以节省通过网络进行通信的费用。然而，芹菜文档没有提供足够的细节，说明如何将结果块称为块。尝试了不同的方法，但它们并不像我预期的那样起作用。我的代码段如下：

@app.task(name='pl.startregret')
def startregret(**kwargs): 
    items = list(zip(range(1000), range(1000)))
    chunk = regretclick.chunks(items, 10)
    print(chunk)
    for c in chunk:
        print(c)


@app.task(name='pl.regretclick')
def regretclick(x,y):
    print('got it.')
    return x + y

我读了一些代码，并认为我的代码块应该是一个生成器。然而，打印输出显示

[2014-10-15 13:12:15,930: WARNING/Worker-2] args
[2014-10-15 13:12:15,931: WARNING/Worker-2] subtask_type
[2014-10-15 13:12:15,931: WARNING/Worker-2] kwargs
[2014-10-15 13:12:15,931: WARNING/Worker-2] immutable
[2014-10-15 13:12:15,931: WARNING/Worker-2] options
[2014-10-15 13:12:15,931: WARNING/Worker-2] task

有什么正确的方法来称呼这些小块吗？

谢谢,

更新:我已经阅读了源代码，并尝试了语块()。现在唯一的问题是使用默认队列而不是在celeryconfig中定义的队列。

Answer 1

考虑这样一个简单的添加任务。

@app.task()
def add(x, y):
    return x + y

下面是一种调用块任务的简单方法。

res = add.chunks(zip(range(10), range(10)), 2)()

这将给定的10个任务块为5个大小为2的任务，并将chunked tasks添加到默认队列中。如果要将其路由到不同的队列，则必须在调用任务时指定它。

res = add.chunks(zip(range(10), range(10)), 2).apply_async(queue='my_special_queue')

然后启动此队列的工作人员以使用任务。

worker -A your_app worker -l info -Q my_special_queue