如何将一系列任务路由到芹菜中的特定队列?

问题描述

当我将任务路由到特定队列时，它会起作用:

When I route a task to a particular queue it works:

task.apply_async(queue='beetroot')

但是如果我创建一个链:

But if I create a chain:

chain = task | task

然后我写

chain.apply_async(queue='beetroot')

它似乎忽略了 queue 关键字并分配给默认的 'celery' 队列.

It seems to ignore the queue keyword and assigns to the default 'celery' queue.

如果 celery 支持链中路由就好了 - 所有任务都在同一个队列中按顺序执行.

It would be nice if celery supported routing in chains - all tasks executed sequentially in the same queue.

推荐答案

好的，我想通了.

您必须将所需的执行选项(如 queue= 或 countdown=)添加到子任务定义中，或者通过部分:

You have to add the required execution options like queue= or countdown= to the subtask definition, or through a partial:

子任务定义:

from celery import subtask

chain = subtask('task', queue = 'beetroot') | subtask('task', queue = 'beetroot')

部分:

chain = task.s().apply_async(queue = 'beetroot') | task.s().apply_async(queue = 'beetroot')

然后你通过以下方式执行链:

Then you execute the chain through:

chain.apply_async()

或者，

chain.delay()

任务将被发送到甜菜根"队列.最后一个命令中的额外执行参数不会做任何事情.在链(或组，或任何其他 Canvas 原语)级别应用所有这些执行参数会很不错.

And the tasks will be sent to the 'beetroot' queue. Extra execution arguments in this last command will not do anything. It would have been kind of nice to apply all of those execution arguments at the Chain (or Group, or any other Canvas primitives) level.