在运行 celery 应用程序中更改并发/管理队列答案

【问题标题】：Change concurrency/manage queue in running celery app在运行 celery 应用程序中更改并发/管理队列
【发布时间】：2021-10-03 12:47:45
【问题描述】：

我正在使用 celery 5 来管理快速 API 应用程序中的一些外部任务。我从一名工人和 8 个并发工作开始 celery：

celery worker --app=app.worker.celery --concurrency=8 --loglevel=info --logfile=logs/celery.log

我希望能够从快速 API 应用更改并发。我不知道这是否是最好的方法，或者即使它是可能的。我还没有找到改变并发的方法，所以我尝试添加新的工作人员，使用

from celery import current_app as app
cmd = ["--app=app.worker.celery", "--concurrency=8", "--loglevel=info", "--logfile=logs/celery.log", "--without-gossip" , "--detach", "-E"]
app.worker_main(cmd)

但即使传递 --detach 也不起作用，它会阻止请求。

还有其他/更好的方法吗？

编辑：

在查看 flower 1.0.1 是如何做到这一点后，我能够跟踪正确的 API。已解决：

from celery import current_app as app
response = app.control.pool_grow(
            n=4, reply=True, destination=[worker_name])

【问题讨论】：

您是否测试过是否可以在命令行（而不是 Python 脚本）中使用这些参数运行 Celery worker？删除 --logfile=logs/celery.log 以便您可以看到工作器启动时出现的任何最终错误。
@DejanLekic，是的，通过 CMD 可以工作，--detach 也可以工作，它会创建一个后台进程。

标签： python celery

【解决方案1】：

您可以通过子类化 celery.worker.autoscale.AutoScaler 并设置 worker_autoscaler 来编写自己的自动缩放器。例如，您的自动缩放器可以监控数据库记录，并在其发生变化时进行扩展。

【讨论】：

嗯，这可能是一个不错的选择。我会尝试。我发现花有一个 API api/worker/pool/grow/ 可以增加一个工人的任务数量.. 试图找到如何直接使用 celery 调用它

【解决方案2】：

Flower 允许增长和缩小池（here 花实现）

from celery import current_app as app
n = 10 # increase the pool size
worker_name = "my_worker"
response = app.control.pool_grow(
            n=n, reply=True, destination=[worker_name])

唯一的问题是它没有在更改池大小后获取池大小的选项

【讨论】：

这应该是一个可以接受的答案。 @Rafael，您从 stats inspect 命令获取池信息。