对广播任务使用芹菜优先级队列

Question

我想在我的芹菜工人中实现任务优先级。我可以通过为高优先级任务和低优先级任务创建不同的队列来实现这一点。但我也需要发送广播任务到所有工人与广播队列，它不工作。给你任务.py文件：

from celery import Celery
from kombu.common import Broadcast, Queue, Exchange

app = Celery('tasks')

app.conf.update(
    CELERY_RESULT_BACKEND='amqp',
    CELERY_ACCEPT_CONTENT=['json'],
    CELERY_TASK_SERIALIZER='json',
    CELERY_RESULT_SERIALIZER='json',
    BROKER_URL='amqp://',
    CELERY_QUEUES=(Queue('default',
                         Exchange('default'),
                         routing_key='default'),
                   Queue('low_priority',
                         Exchange('low_priority'),
                         routing_key='low_priority'),
                   Broadcast('broadcast_tasks'), ),
    CELERY_ROUTES={'tasks.broadcast':
                   {'queue': 'broadcast_tasks'},
                   'tasks.low_task':
                   {'queue': 'low_priority'},
                   },
    CELERY_DEFAULT_QUEUE = 'default',
    CELERY_DEFAULT_EXCHANGE = 'default',
    CELERY_DEFAULT_ROUTING_KEY = 'default'
)

@app.task
def broadcast():
    print "Broadcast called"

@app.task
def low_task():
    print "Low priority called"

@app.task
def def_task():
    print "Default called"

当我用这个命令运行芹菜工人时：

^{pr2}$

任务优先级有效，但广播任务未被确认。在

当我在没有队列参数的情况下运行同一命令时，广播可以工作，但任务优先级不起作用：

celery -A tasks worker --loglevel=info
celery -A tasks worker --loglevel=info

据我所知，发生这种情况是因为广播队列具有唯一的名称，例如每个工作线程的bcast.0b5dbce0-9bcb-48a5-8554-cbb7f32a6703。在

有人有好的解决办法吗？提前谢谢！在

Answer 1

您必须显式地使用广播队列，因此根据ANDY_VAR将命令行调用修改为上面的注释

这里提出了一个类似的问题：

• start celery worker and enable it for broadcast queue