Как компоновать структуру очереди/работника для поддержки больших задач для нескольких сред?

Для инструмента развертывания на основе Python/Django/Celery у нас есть следующая настройка:

  • В настоящее время мы используем настройку по умолчанию Celery. (Одна очередь + обмен называется "сельдерей".)
  • Каждая задача в очереди представляет собой операцию развертывания.
  • Каждая задача для среды заканчивается фазой синхронизации, которая потенциально занимает (очень) длинную.

Следующие спецификации должны быть выполнены:

  • Concurrency: задачи для нескольких сред должны выполняться одновременно.
  • Блокировка. Одновременно может выполняться не более одной задачи для каждой среды (т.е. блокировка среды).
  • Оптимизация пропускной способности. При наличии нескольких задач для одной среды их фазы синхронизации могут быть объединены для оптимизации. Поэтому, если задача приближается к ее завершению, она должна проверить, ждут ли в очереди очередные задачи для этой среды, и если да, то пропустите фазу синхронизации.

Каков предпочтительный способ реализации этого?

Некоторые мысли:

  • Я бы сказал, что мы должны настроить несколько очередей: по одному для каждой среды, и N рабочих сельдерея обрабатывает одну очередь исключительно, каждый. (Это решило бы спецификацию 1 + 2.)
    Но как мы можем заставить нескольких работников сельдерея слушать только отдельные очереди?
  • Есть ли чистый способ узнать, есть ли в очереди очередные задачи для среды?

Ответы

Ответ 1

для 1,2 используйте несколько очередей и запускающих работников с -Q, чтобы указать, какую очередь прослушать. Также настройте CELERYD_PREFETCH_MULTIPLIER = 1 только для одной задачи за раз.

Чтобы получить длину очереди (проверено с помощью rabbitmq), вы можете использовать что-то вроде этого:

from kombu.connection import BrokerConnection
connection = BrokerConnection(BROKER_HOST, BROKER_USER...)
channel = connection.channel()
q, j, c = channel.queue_declare('celery', passive=True)
print 'celery %d jobs in queue' % j

'queue_delcare' как побочный эффект, дайте вам длину очереди. Надеюсь, это поможет вам.

Ответ 2

Я бы посмотрел zeromq, он может делать обмен сообщениями и многопоточность в одной супер быстрой библиотеке. Он также поддерживает большое количество языков и имеет встроенную балансировку нагрузки.