python - 如何使用 Celery 从 Python 代码动态添加/停止 worker

Question

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

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

我尝试在启动时从我的 Django 项目中启动 worker。但是我好像不止一个启动失败。我最初尝试在我的 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 源代码创建 worker？

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

Answer 1

您拥有的代码是创建新工作程序的合法方式。您可能还想调查在启动时分离工作人员。也就是说，如果您有很多用户，此实现会受到两个事实的限制:(1) 所有工作人员都在同一台机器上运行，以及 (2) 调用者无法控制工作人员停止的方式/时间。