Как динамически добавлять / останавливать работников из кода Python с помощью Celery

Question

Я работаю с докернизированной системой Django, которая обрабатывает буферы для пользователей. Допустим, у меня 3 пользователя, и один из них решает поставить в очередь тысячи буферов. Я не хочу, чтобы другие 2 подождали, пока все они не будут завершены.

Можно ли динамически запустить работника, который обрабатывает только те задачи, которые находятся в очереди, назначенной этому пользователю? Я знаю, что вы можете динамически определять очередь, в которую должна быть помещена задача, после ее вызова.

Я попытался запустить рабочих из моего проекта Django при запуске. Но я, кажется, не могу начать больше, чем просто один. Сначала я попытался сделать это в своем конфиге сельдерея:

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?

Если это не тот путь, как я могу динамически создавать или удалять рабочих для моих пользователей?

Answer 1

Код, который у вас есть, законный способ создать нового работника. Возможно, вы также захотите исследовать возможность отсоединения работника при его запуске. Тем не менее, если у вас много пользователей, эта реализация ограничена двумя фактами (1) все работники работают на одной машине и (2) вызывающая сторона не контролирует, как / когда работник остановится.