python - 带有redis的 celery 中的任务优先级

Question

我想用celery实现一个分布式作业执行系统。鉴于 rabbitMQ 不支持优先级，而我非常需要这个功能，我转向了 celery+redis。

在我的情况下，任务与硬件密切相关，例如，任务 A 只能在 Worker 1 上运行，因为只有 Worker 1 的 PC 有必要的硬件。我将每个 worker 的CONCURRENCY设置为1，这样一个 worker 每次只会运行一个任务。每个任务大约需要 2 分钟。

为了实现优先级功能，首先我尝试在调用apply_async()时添加priority参数，例如apply_async(priority=0) 和 apply_async(priority=9)。在这个测试中，我只启动了一个COCURRENCY=1的Worker，并分别启动了10个不同优先级的任务。我希望看到 apply_async(priority=0) 启动的任务将优先运行，但不幸的是它们只是作为启动顺序启动。

然后我尝试做一些变通。我克隆了每个任务，所以对于每个任务，我都有 task_high 和 task_low，由 @celery.task(priority=0) 和 @celery.task(priority=1) 修饰.然后我做了和上面一样的测试，这次更好，当开球顺序是“HH-LLLL-HHHH”时，真正的顺序是“HH-L-H-H-L-H-L-L-H”。我想 Redis 在这里做了一些调度和平衡工作。

但这还是达不到我的期望。我希望得到像“HHHHHH-LLLL”这样的命令，因为对于某些任务，我只有一台合适的机器和必要的硬件，希望高优先级的任务尽快运行。

我在网上搜索过其他的解决方法，比如用两个队列，一个用于高优先级任务，一个用于低优先级任务，前者使用2台机器，后者使用1台机器。但由于我的硬件非常有限，这对我不起作用。

你能给点建议吗？

Answer 1

Celery Redis 传输确实尊重优先级字段， 但是 Redis 本身没有优先级的概念。

通过为每个队列创建n个列表来实现优先级支持 并在 BRPOP 命令中使用该命令。 我在这里说 n 是因为即使有 10 (0-9) 个优先级，这些都是 默认合并为 4 个级别以节省资源。 这意味着名为 celery 的队列实际上将被分成 4 个队列:

['celery0', 'celery3`, `celery6`, `celery9`]

如果你想要更多的优先级，你可以设置 priority_steps 传输选项:

BROKER_TRANSPORT_OPTIONS = {
    'priority_steps': list(range(10)),
}

也就是说，请注意，这永远不会像实现的优先事项那样好 在服务器级别，最多可能是近似值。但它可能仍然很好 足以满足您的应用需求。

Source.