如何使用 Celery 从 Python 代码中动态添加/停止工作人员答案

【问题标题】：How to dynamically add/stop workers from Python code with Celery如何使用 Celery 从 Python 代码中动态添加/停止工作人员
【发布时间】：2019-05-29 04:55:23
【问题描述】：

我正在使用一个为用户处理缓冲区的 dockerized Django 系统。假设我有 3 个用户，其中一个决定将数千个缓冲区排队。我不希望其他两个必须等待所有这些都完成。

是否可以动态启动仅处理分配给该用户的队列中的任务的工作人员？我知道你可以在调用任务时动态决定任务应该放在哪个队列中。

我尝试在启动时从我的 Django 项目中启动工作人员。但我似乎无法开始不止一个。我最初尝试在我的 celery 配置中这样做：

import os

from celery import Celery
from celery.bin import worker
from django.conf import settings

os.environ.setdefault("DJANGO_SETTINGS_MODULE", "config.settings")

app = Celery('project1')

app.config_from_object('django.conf:settings')

app.autodiscover_tasks(lambda: settings.INSTALLED_APPS)

app.conf.ONCE = {
  'settings': {
    'url': 'redis://' + 'redis',
    'blocking': True,
    'default_timeout': 60 * 60,
    'blocking_timeout': 86400
  }
}

print("Starting worker...")

for e in Customer.objects.all():

  worker = worker.worker(app=app)
  worker.run(queues=["Queue" + e.id.__str__(),])

这甚至是一种合法的方式吗？如果是这样，我如何从我的 Python 源代码创建工作者？

如果不是这样，我如何为我的用户动态创建或删除工作人员？

【问题讨论】：

标签： python django docker celery

【解决方案1】：

您拥有的代码是一种创建新工作者的合法方式。您可能还想研究在启动它时将其分离。也就是说，如果您有很多用户，则此实现受到两个事实的限制：（1）所有工作人员都在同一台机器上运行，（2）调用者无法控制工作人员停止的方式/时间。

【讨论】：