芹菜任务优先级

Question

我想使用 Celery 管理任务。我想要一个单一的任务队列（并发度为 1），并且能够将任务推送到具有不同优先级的队列中，以便更高优先级的任务将抢占其他任务。

我正在向队列中添加三个任务，如下所示：

add_tasks.py

from tasks import example_task

example_task.apply_async((1), priority=1)
example_task.apply_async((2), priority=3)
example_task.apply_async((3), priority=2)

我有以下配置：

tasks.py

from __future__ import absolute_import, unicode_literals
from celery import Celery
from kombu import Queue, Exchange
import time

app = Celery('tasks', backend='rpc://', broker='pyamqp://')

app.conf.task_queues = [Queue('celery', Exchange('celery'), routing_key='celery', queue_arguments={'x-max-priority': 10})]


@app.task
def example_task(task_num):
    time.sleep(3)
    print('Started {}'.format(task_num)
    return True

我希望我在第三个之前添加到运行的第二个任务，因为它具有更高的优先级，但它没有。他们运行按顺序添加。

我正在关注文档并认为我已经正确配置了应用程序。

我是做错了什么还是误解了优先功能？

Answer 1

队列有可能没有机会确定消息的优先级（因为它们在排序发生之前就已下载）。尝试使用这两个设置（根据需要适应您的项目）：

CELERY_ACKS_LATE = True
CELERYD_PREFETCH_MULTIPLIER = 1

预取倍数默认为4。

我开发了一个示例应用程序来在非常小的范围内实现 Celery 的优先任务。请看一下here。在开发它的时候，我遇到了一个非常相似的问题，这个设置的改变实际上解决了它。

请注意，您还需要 RabbitMQ 版本 3.5.0 或更高版本。

芹菜任务优先级

Celery Task Priority

python

celery