如何在 celery task.apply_async 中使用优先级

2022-01-11 00:00:00 python queue celery rabbitmq priority-queue

问题描述

我在 celery 中有一个 test 队列，我已经为它定义了一个任务:

I have a test queue in celery and I have defined a task for it:

@celery_app.task(queue='test', ignore_result=True)
def priority_test(priority):
    print(priority)

只打印参数.我想设置 priority 属性appy_async 的 .task.html#celery.app.task.Task.apply_async" rel="noreferrer">这里.所以，我像这样写了一个 for 循环:

which just print the argument.I want to set the priority attribute which is defined here for appy_async. So, I wrote a for loop like this:

for i in range(100):
    priority_test.apply_async((i%10,), queue="test", priority=i%10)

我希望看到这样的结果:

I excpected to see some result like this:

[2017-12-26 17:21:37,309: WARNING/ForkPoolWorker-1] 10
[2017-12-26 17:21:37,311: WARNING/ForkPoolWorker-1] 10
[2017-12-26 17:21:37,314: WARNING/ForkPoolWorker-1] 10
[2017-12-26 17:21:37,317: WARNING/ForkPoolWorker-1] 9
[2017-12-26 17:21:37,319: WARNING/ForkPoolWorker-1] 9
[2017-12-26 17:21:37,321: WARNING/ForkPoolWorker-1] 9
[2017-12-26 17:21:37,323: WARNING/ForkPoolWorker-1] 8
[2017-12-26 17:21:37,326: WARNING/ForkPoolWorker-1] 8
[2017-12-26 17:21:37,329: WARNING/ForkPoolWorker-1] 8
[2017-12-26 17:21:37,332: WARNING/ForkPoolWorker-1] 7
[2017-12-26 17:21:37,334: WARNING/ForkPoolWorker-1] 7
[2017-12-26 17:21:37,336: WARNING/ForkPoolWorker-1] 7
[2017-12-26 17:21:37,341: WARNING/ForkPoolWorker-1] 6
[2017-12-26 17:21:37,344: WARNING/ForkPoolWorker-1] 6
[2017-12-26 17:21:37,346: WARNING/ForkPoolWorker-1] 6
[2017-12-26 17:21:37,349: WARNING/ForkPoolWorker-1] 5
[2017-12-26 17:21:37,351: WARNING/ForkPoolWorker-1] 5
[2017-12-26 17:21:37,353: WARNING/ForkPoolWorker-1] 5
[2017-12-26 17:21:37,355: WARNING/ForkPoolWorker-1] 4
[2017-12-26 17:21:37,358: WARNING/ForkPoolWorker-1] 4
[2017-12-26 17:21:37,360: WARNING/ForkPoolWorker-1] 4

表示依次执行相同的优先级，但以正常方式执行:

means execute the same priorities after each other but it executed them in the normal way:

[2017-12-26 17:21:37,309: WARNING/ForkPoolWorker-1] 10
[2017-12-26 17:21:37,311: WARNING/ForkPoolWorker-1] 9
[2017-12-26 17:21:37,314: WARNING/ForkPoolWorker-1] 8
[2017-12-26 17:21:37,317: WARNING/ForkPoolWorker-1] 7
[2017-12-26 17:21:37,319: WARNING/ForkPoolWorker-1] 6
[2017-12-26 17:21:37,321: WARNING/ForkPoolWorker-1] 5
[2017-12-26 17:21:37,323: WARNING/ForkPoolWorker-1] 4
[2017-12-26 17:21:37,326: WARNING/ForkPoolWorker-1] 3
[2017-12-26 17:21:37,329: WARNING/ForkPoolWorker-1] 2
[2017-12-26 17:21:37,332: WARNING/ForkPoolWorker-1] 1
[2017-12-26 17:21:37,334: WARNING/ForkPoolWorker-1] 10
[2017-12-26 17:21:37,336: WARNING/ForkPoolWorker-1] 9
[2017-12-26 17:21:37,341: WARNING/ForkPoolWorker-1] 8
[2017-12-26 17:21:37,344: WARNING/ForkPoolWorker-1] 7
[2017-12-26 17:21:37,346: WARNING/ForkPoolWorker-1] 6
[2017-12-26 17:21:37,349: WARNING/ForkPoolWorker-1] 5
[2017-12-26 17:21:37,351: WARNING/ForkPoolWorker-1] 4
[2017-12-26 17:21:37,353: WARNING/ForkPoolWorker-1] 3
[2017-12-26 17:21:37,355: WARNING/ForkPoolWorker-1] 2
[2017-12-26 17:21:37,358: WARNING/ForkPoolWorker-1] 1
[2017-12-26 17:21:37,360: WARNING/ForkPoolWorker-1] 10
[2017-12-26 17:21:37,362: WARNING/ForkPoolWorker-1] 9
[2017-12-26 17:21:37,364: WARNING/ForkPoolWorker-1] 8
[2017-12-26 17:21:37,365: WARNING/ForkPoolWorker-1] 7
[2017-12-26 17:21:37,367: WARNING/ForkPoolWorker-1] 6
[2017-12-26 17:21:37,369: WARNING/ForkPoolWorker-1] 5
[2017-12-26 17:21:37,371: WARNING/ForkPoolWorker-1] 4
[2017-12-26 17:21:37,373: WARNING/ForkPoolWorker-1] 3
[2017-12-26 17:21:37,374: WARNING/ForkPoolWorker-1] 2
[2017-12-26 17:21:37,376: WARNING/ForkPoolWorker-1] 1

我应该如何使用 rabbitmq 在 celery 中应用 priority 以及上面文档中的 priority 属性是什么?

How should I apply priority in celery with rabbitmq and what is the priority attribute in the doc above?

解决方案

为了使 priority 正常工作，您需要正确配置几个设置，并且至少需要 3.5.0 版本RabbitMQ.

In order to have priority working properly you need to properly configure a couple of settings and you need at least version 3.5.0 of RabbitMQ.

首先将队列的 x-max-priority 设置为 10.来自 文档:

First set the x-max-priority of your queue to 10. From the docs :

from kombu import Exchange, Queue

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

可以使用 task_queue_max_priority 设置为所有队列设置默认值:

A default value for all queues can be set using the task_queue_max_priority setting:

app.conf.task_queue_max_priority = 10

然后配置以下设置:

CELERY_ACKS_LATE = True
CELERYD_PREFETCH_MULTIPLIER = 1

默认情况下，预取乘数为 4，在您的情况下，这将导致优先级为 10、9、8 和 7 的前 4 个任务在队列中出现其他任务之前被提取.CELERY_ACKS_LATE 设置将导致任务在执行后得到确认.您可以尝试使用此设置，看看您喜欢哪种行为.

By default the prefetch multiplier is 4, which in your case will cause the first 4 tasks with priority 10, 9, 8 and 7 to be fetched before the other tasks are present in the queue. The CELERY_ACKS_LATE setting will cause the tasks to be acknowledged after they have been executed. You can experiment with this setting to see what behaviour you prefer.

相关文章