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

Question

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

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

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

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

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

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

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

Answer 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' в качестве побочного эффекта, даст вам длину очереди. Надеюсь, это поможет вам.

Answer 2

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