python - Celery 使用一批消息执行任务

Question

我想向 celery 发送消息，当它达到 100 条消息时，我希望 celery 分批执行它们。如果我想批量提交到数据库，这是一种常见的情况。

为了这个目的，在谷歌搜索时我发现了这个链接:用 celery 做批处理: http://celery.readthedocs.org/en/latest/reference/celery.contrib.batches.html

我的问题是在示例中没有明显的方法来获取提交给任务的数据

例如，假设我们通过以下方式逐条提交一些消息:

task.apply_async((message,), link_error=error_handler.s())

然后我们有以下任务实现:

@celery.task(name="process.data", base=Batches, flush_every=100, flush_interval=1)
def process_messages(requests):
   for request in requests:
       print request /// how I can take the message data submitted in my task for process?

有没有其他方法可以用 celery 实现批处理？谢谢

Answer 1

对于那些在多次尝试和错误后发现这篇文章有用的人，我已经设法通过以下方式从 SimplRequest 对象中取出数据:

当您通过以下方式提交数据时:

func.delay(data)

从请求对象中，您可以获得 args 属性，它是一个包含数据的列表:

request.args[0]
request.args[1] 
etc.

如果您通过以下方式提交您的数据:

func.apply_async((), {'data': data}, link_error=error_handler.s())

然后数据在 kwargs 中作为字典可用:

request.kwargs['data']

最后，如示例所示，我们需要对所有请求进行循环以收集数据批处理

for r in requests:
       data = r.kwargs['data']

文档页面 ( here ) 中的示例最好能更新为更简单明了的示例