Как узнать, когда набор задач RabbitMQ завершен?

Question

Я использую RabbitMQ, чтобы рабочие процессы кодировали видеофайлы.Я хотел бы знать, когда все файлы будут завершены, то есть когда все рабочие процессы будут завершены.

Единственный способ, которым я могу думать, это использовать базу данных.Когда видео завершает кодирование:

UPDATE videos SET status = 'complete' WHERE filename = 'foo.wmv'
-- etc etc etc as each worker finishes --

И затем, чтобы проверить, все ли видео были закодированы:

SELECT count(*) FROM videos WHERE status != 'complete'

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

Существует ли стандартный механизм для зависимостей RabbitMQ?То есть способ сказать «подождать, пока эти 5 задач завершатся, и как только они будут выполнены, а затем запустить новую задачу?»

Я не хочу, чтобы родительский процесс добавил эти задачи вочередь, а затем «ждать», пока каждый из них вернет «завершенный» статус.Затем я должен поддерживать отдельный процесс для каждой группы видео, и в этот момент я потерял преимущество отсоединенных рабочих процессов по сравнению с одной концепцией ThreadPool.

Я спрашиваю о том, что невозможно?Или существуют стандартные широко распространенные решения для управления общим состоянием задач в очереди, которые я пропустил?

Редактировать: после поиска я нашел такой похожий вопрос: Получение результата длительного выполнениязадание с RabbitMQ

Есть ли у людей какие-то особые мысли по этому поводу?

Answer 1

Использовать очередь «ответа». Я не знаю никаких подробностей о RabbitMQ, так что это общее:

• Ваш родительский процесс отправляет запросы и отслеживает, сколько он отправил
• Заставить родительский процесс также ждать в определенной очереди ответов (о которой знают дети)
• Всякий раз, когда ребенок что-то заканчивает (или по какой-то причине не может закончить), отправляет сообщение в очередь ответов
• Всякий раз, когда numSent == numResponded, вы сделали

Нужно иметь в виду, что это тайм-аут - что произойдет, если умирает дочерний процесс? Вы должны сделать немного больше работы, но в основном:

• В каждом отправленном сообщении укажите какой-нибудь идентификатор и добавьте этот идентификатор и текущее время в хеш-таблицу.
• Для каждого ответа удаляйте этот идентификатор из хеш-таблицы
• Периодически обходите хеш-таблицу и удаляйте все, что истекло

Это называется Шаблон ответа на запрос .

Answer 2

Я реализовал рабочий процесс, в котором конечный автомат рабочего процесса реализован в виде серии очередей. Рабочий получает сообщение в одну очередь, обрабатывает работу, а затем публикует то же сообщение в другой очереди. Затем другой тип рабочего процесса получает это сообщение и т. Д.

В вашем случае кажется, что вам нужно реализовать один из шаблонов из Шаблонов корпоративной интеграции (это бесплатная онлайн-книга) и иметь простого работника, который собирает сообщения, пока набор работ готово, а затем обрабатывает одно сообщение в очередь, представляющую следующий шаг в рабочем процессе.

Answer 3

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