python - celery 一个经纪人多个队列和 worker

Question

我有一个名为 tasks.py 的 python 文件，我在其中定义了 4 个单一任务。我想配置 celery 以使用 4 个队列，因为每个队列都会分配不同数量的工作人员。我正在阅读我应该使用 route_task 属性，但我尝试了几种选择但没有成功。

我正在关注这个文档 celery route_tasks docs

我的目标是运行 4 个工作器，每个工作器一个，不要将来自不同队列中的不同工作器的任务混在一起。这是可能的？这是一个好方法吗？

如果我做错了什么，我很乐意更改我的代码以使其正常工作

这是我目前的配置

任务.py

app = Celery('tasks', broker='pyamqp://guest@localhost//')
app.conf.task_default_queue = 'default'
app.conf.task_queues = (
    Queue('queueA',    routing_key='tasks.task_1'),
    Queue('queueB',    routing_key='tasks.task_2'),
    Queue('queueC',    routing_key='tasks.task_3'),
    Queue('queueD',    routing_key='tasks.task_4')
)


@app.task
def task_1():
    print "Task of level 1"


@app.task
def task_2():
    print "Task of level 2"


@app.task
def task_3():
    print "Task of level 3"


@app.task
def task_4():
    print "Task of level 4"

为每个队列运行一个celery worker

celery -A tasks worker --loglevel=debug -Q queueA --logfile=celery-A.log -n W1&
celery -A tasks worker --loglevel=debug -Q queueB --logfile=celery-B.log -n W2&
celery -A tasks worker --loglevel=debug -Q queueC --logfile=celery-C.log -n W3&
celery -A tasks worker --loglevel=debug -Q queueD --logfile=celery-D.log -n W4&

Answer 1

无需为将任务提交到不同队列而进行复杂的路由。像往常一样定义您的任务。

from celery import celery

app = Celery('tasks', broker='pyamqp://guest@localhost//')

@app.task
def task_1():
    print "Task of level 1"


@app.task
def task_2():
    print "Task of level 2"

现在在对任务进行排队时，将任务放入适当的队列中。这是有关如何执行此操作的示例。

In [12]: from tasks import *

In [14]: result = task_1.apply_async(queue='queueA')

In [15]: result = task_2.apply_async(queue='queueB')

这会将 task_1 放入名为 queueA 的队列中，并将 task_2 放入 queueB 中。

现在您可以让您的工作人员开始使用它们。

celery -A tasks worker --loglevel=debug -Q queueA --logfile=celery-A.log -n W1&
celery -A tasks worker --loglevel=debug -Q queueB --logfile=celery-B.log -n W2&