Jak sprawdzić, kiedy zestaw zadań RabbitMQ jest kompletny?

Question

Używam RabbitMQ, aby procesy robocze kodowały pliki wideo. Chciałbym wiedzieć, kiedy wszystkie pliki są kompletne - to znaczy, gdy wszystkie procesy robocze zostały zakończone.

Jedyny sposób, w jaki mogę to zrobić, to użycie bazy danych. Po zakończeniu kodowania wideo:

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

A następnie sprawdzić, czy wszystkie filmy zostały zakodowane:

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

Ale jeśli mam zamiar to zrobić, wtedy czuję się jak tracę korzyści RabbitMQ jako mechanizmu dla wielu rozproszonych procesów roboczych, ponieważ wciąż muszę ręcznie utrzymywać kolejkę bazy danych.

Czy istnieje standardowy mechanizm dla zależności RabbitMQ? Oznacza to, że można powiedzieć "poczekaj aż te 5 zadań się zakończy, a kiedy skończą, a następnie rozpoczną nowe zadanie?"

Nie chcę, aby proces nadrzędny dodał te zadania do kolejki, a następnie "zaczekaj", aż każdy z nich zwróci status "ukończony". Następnie muszę utrzymywać oddzielny proces dla każdej grupy filmów, w którym to momencie straciłem przewagę oddzielonych procesów roboczych w porównaniu z pojedynczą koncepcją ThreadPool.

Czy proszę o coś, co jest niemożliwe? Czy istnieją standardowe, powszechnie przyjęte rozwiązania do zarządzania ogólnym stanem zadań w kolejce, które przeoczyłem?

Edycja: po poszukiwaniach, znalazłem to podobne pytanie: Getting result of a long running task with RabbitMQ

istnieją jakieś szczególne myśli, że ludzie mają na ten temat?

Answer 1

Użyj kolejki "odpowiedzi". Nie znam żadnych szczegółów dotyczących RabbitMQ, więc jest to ogólnie:

• Czy Twój proces nadrzędny wysyła żądania i śledzić, ile on wysłany
• uczynić proces rodzic również czekać na konkretnej kolejce odpowiedzi (że dzieci wiedzą o)
• Kiedy dziecko kończy coś (lub nie może skończyć z jakiegoś powodu), wysłać wiadomość do kolejki odpowiedzi
• Ilekroć numSent == numResponded, skończysz

Należy pamiętać o limicie czasu - co stanie się, gdy proces potomny umrze? Trzeba wykonać nieco więcej pracy, ale zasadniczo:

• Przy każdej wysłanej wiadomości należy podać identyfikator i dodać ten identyfikator oraz aktualny czas do tablicy mieszającej.
• Dla każdej odpowiedzi usunąć ten identyfikator z tabeli hash
• Okresowo chodzić tabeli mieszania i usunąć wszystko, co wygasła

To się nazywa Request Reply Pattern.

Answer 2

Zaimplementowałem przepływ pracy, w którym maszyna stanu przepływu pracy jest zaimplementowana jako seria kolejek.Pracownik odbiera komunikat w jednej kolejce, przetwarza pracę, a następnie publikuje ten sam komunikat w innej kolejce. Następnie inny typ procesu roboczego odbiera tę wiadomość, itp.

W twoim przypadku wygląda na to, że musisz zaimplementować jeden z szablonów z Enterprise Integration Patterns (jest to darmowa książka online) i mieć prostego pracownika, który zbiera wiadomości do wykonania zestawu zadań, a następnie przetworzenia pojedynczej wiadomości do kolejki reprezentującej kolejny etap przepływu pracy.