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

Для инструмента развертывания на основе 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, он может делать обмен сообщениями и многопоточность в одной супер быстрой библиотеке. Он также поддерживает большое количество языков и имеет встроенную балансировку нагрузки.