Kolejki w Java i mikroserwisach – ActiveMQ, RabbitMQ, SQS i SNS

Czym są kolejki?

Kolejka to struktura danych, która przechowuje elementy w określonej kolejności i pozwala pobierać je zgodnie z przyjętą zasadą, najczęściej FIFO (First In, First Out). Oznacza to, że pierwszy element dodany do niej jest również pierwszym, który zostanie z niej pobrany.

W aplikacjach kolejki są wykorzystywane nie tylko jako zwykłe struktury danych, ale również jako sposób organizowania zadań i wymiany wiadomości pomiędzy różnymi częściami systemu. Do komunikacji z kolejką przeważnie stosuje się dwa rodzaje aplikacji. Jedną, która jest producentem, czyli dodaje wiadomość do kolejki, oraz drugą – konsument – która pobiera ją i przetwarza.

Jakie problemy rozwiązuje kolejka?

Kolejka to jeden z podstawowych mechanizmów wykorzystywanych do organizowania i przetwarzania zadań w aplikacjach. Pozwalają oddzielić moment utworzenia zadania od momentu jego wykonania, dzięki czemu producent nie musi czekać, aż odbiorca zakończy przetwarzanie. Jest to szczególnie przydatne wtedy, gdy operacja przetwarzania trwa długo. Może wtedy ona zostać wykonana w tle albo zostać chwilowo wstrzymana, gdy przekracza możliwości systemu. Kolejka pomaga więc kontrolować obciążenie systemu, umożliwia przetwarzanie danych w odpowiednim tempie i ułatwia budowanie komunikacji asynchronicznej (czyli nadawca wysyła wiadomość bez czekania na jego natychmiastowe wykonanie). W prostych aplikacjach może działać bezpośrednio w pamięci JVM, natomiast w systemach rozproszonych i mikroserwisach tę samą ideę realizują zewnętrzne brokery wiadomości oraz usługi chmurowe.

Kolejki dostępne w Java JDK

Java udostępnia kilka implementacji kolejek w ramach standardowego JDK. Podstawą jest interfejs Queue, a do najczęściej używanych implementacji należą między innymi ArrayDeque, LinkedList oraz PriorityQueue. Typowe operacje wykonywane są za pomocą metod offer(), poll() i peek().

Nie będę tutaj szczegółowo omawiał ich działania, ponieważ zrobiłem to już w osobnym wpisie odnośnie BlockingQueue. Jeśli chcesz przypomnieć sobie podstawowe operacje na kolejkach w Java, zajrzyj do artykułu Java #59: struktury danych bezpieczne wątkowo

W tym wpisie skupię się przede wszystkim na tym, jak wykorzystać kolejki do przetwarzania zadań oraz komunikacji asynchronicznej.

Message Broker – kolejka poza aplikacją

Kolejki dostępne bezpośrednio w Java działają wewnątrz pamięci konkretnej aplikacji. Jest to wystarczające, jeśli producent i konsument działają w ramach jednej JVM. Problem pojawia się jednak wtedy, gdy wiadomości mają być przekazywane pomiędzy niezależnymi aplikacjami lub mikroserwisami.

W takim przypadku można wykorzystać Message Broker, czyli zewnętrzny system pośredniczący w wymianie wiadomości. Producent nie wysyła komunikatu bezpośrednio do konsumenta, lecz przekazuje go do brokera:

Producer -> Message Broker -> Consumer

Broker przechowuje wiadomość do momentu, w którym konsument będzie mógł ją odebrać i przetworzyć. Dzięki temu producent i konsument nie muszą działać jednocześnie. Jeśli konsument jest chwilowo niedostępny, wiadomość może pozostać w kolejce i zostać obsłużona później.

Takie rozwiązanie daje kilka istotnych korzyści. Przede wszystkim zmniejsza zależność pomiędzy aplikacjami, umożliwia komunikację asynchroniczną i pozwala lepiej radzić sobie z chwilowymi skokami obciążenia. Zewnętrzne brokery oferują również mechanizmy takie jak potwierdzanie przetworzenia wiadomości, ponowne próby dostarczenia czy Dead Letter Queue dla komunikatów, których nie udało się poprawnie obsłużyć.

Dead Lette Queue to osobna kolejka przeznaczona na wiadomości, których nie udało się poprawnie przetworzyć mimo kolejnych prób (np. po to aby później je zdiagnozować).

Do popularnych brokerów wiadomości należą między innymi Apache ActiveMQ oraz RabbitMQ. Ich konkretne możliwości różnią się, ale podstawowa idea pozostaje taka sama – kolejka nie znajduje się już wewnątrz pojedynczej aplikacji, lecz staje się niezależnym elementem infrastruktury systemu.

Kolejki w architekturze mikroserwisowej

W architekturze mikroserwisowej kolejki są często wykorzystywane do komunikacji pomiędzy niezależnymi usługami. Zamiast wywoływać inny mikroserwis bezpośrednio i czekać na jego odpowiedź, aplikacja może wysłać wiadomość do kolejki, która zostanie przetworzona później przez odpowiedniego konsumenta.

Przykładowo OrderService po utworzeniu zamówienia może wysłać komunikat OrderCreated do kolejki. NotificationService lub inny mikroserwis może następnie odebrać odpowiednią wiadomość i wykonać swoją część pracy.

Takie podejście zmniejsza zależności pomiędzy mikroserwisami. Producent nie musi wiedzieć, gdzie znajduje się konsument ani czy jest on w danym momencie dostępny. Jeśli odbiorca chwilowo nie działa, wiadomość może pozostać w kolejce i zostać przetworzona po jego ponownym uruchomieniu.

Trzeba jednak pamiętać, że komunikacja przez kolejkę jest bardziej złożona niż standardowa komunikacja poprzez REST (np. GET lub POST). Należy uwzględnić między innymi możliwość ponownego dostarczenia wiadomości, błędy przetwarzania, kolejność komunikatów oraz obsługę wiadomości, których nie udało się poprawnie przetworzyć.

Duplikaty i idempotencja

W systemach kolejkowych może wystąpić taka sytuacja, że ta sama wiadomość zostanie dostarczona więcej jeden raz.

Dlatego konsumenci powinni być idempotentni – ponowne przetworzenie tej samej wiadomości nie powinno powodować ponownego wykonania operacji biznesowej.

Popularnym rozwiązaniem jest nadawanie wiadomościom unikalnego identyfikatora:

{
"eventId": "912ea3e7-54db-4df5-a25c-127d1f784b82",
"eventType": "OrderCreated",
"orderId": "12345"
}

Konsument może sprawdzić, czy dane eventId zostało już wcześniej przetworzone.

Mechanizmy retry

W systemach opartych na kolejkach bardzo często stosuje się mechanizm retry, czyli ponowną próbę przetworzenia wiadomości, gdy nastąpi błąd. W ekosystemie Spring można taki mechanizm zrealizować na kilka sposobów, na przykład przy pomocy Spring Retry albo komponentów Spring Integration. Retry może wykonywać kolejne próby natychmiast lub z opóźnieniem, często z wykorzystaniem strategii backoff, czyli stopniowego zwiększania czasu pomiędzy próbami. Ważne jest jednak, aby liczba ponowień była ograniczona.

@Service
public class MessageProcessor {

    @Retryable(
        retryFor = RuntimeException.class,
        maxAttempts = 3,
        backoff = @Backoff(delay = 1000, multiplier = 2)
    )
    public void process(Message message) {
        // przetwarzanie wiadomości
        externalService.send(message);
    }

    @Recover
    public void recover(RuntimeException exception, Message message) {
        // po wyczerpaniu retry wiadomość może trafić do DLQ
        deadLetterQueue.send(message);
    }
}

Komunikacja asynchroniczna w AWS – SQS i SNS

Chmura AWS oferuje kilka usług przeznaczonych do komunikacji asynchronicznej. Dwie z najczęściej wykorzystywanych to Amazon SQS (Simple Queue Service) oraz Amazon SNS (Simple Notification Service).

SQS to klasyczna kolejka. Producent wysyła wiadomość do kolejki, gdzie może ona oczekiwać do momentu pobrania i przetworzenia przez konsumenta. Dzięki temu producent i konsument mogą działać niezależnie od siebie.

Producer -> SQS Queue -> Consumer

SNS działa natomiast w modelu Publish/Subscribe. Producent publikuje wiadomość do tzw. Topic, a SNS przekazuje ją do wszystkich subskrybentów zainteresowanych danym zdarzeniem.

                   -> Subscriber A
Producer -> SNS    -> Subscriber B
                   -> Subscriber C

SQS i SNS bardzo często stosuje się razem. Popularnym rozwiązaniem jest wzorzec Fan-out, w którym SNS otrzymuje jedno zdarzenie, a następnie przekazuje jego kopię do kilku niezależnych kolejek SQS:

                       -> SQS -> PaymentService
                      /
OrderService -> SNS --+--> SQS -> WarehouseService
                      \
                       -> SQS -> NotificationService

Każdy mikroserwis posiada wtedy własną kolejkę i może przetwarzać wiadomości we własnym tempie. Jeśli jeden z konsumentów jest chwilowo niedostępny, nie wpływa to na pozostałe usługi, a jego wiadomości pozostają w odpowiedniej kolejce SQS do późniejszego przetworzenia.

Najczęstsze problemy i dobre praktyki

Kolejki rozwiązują wiele problemów związanych z komunikacją pomiędzy usługami, ale jednocześnie wprowadzają własne wyzwania. Warto brać je pod uwagę już podczas projektowania systemu. O to kilka z nich (część z nich już omawialiśmy).

Retry i Dead Letter Queue

Nie każdy błąd oznacza, że wiadomości nie da się przetworzyć. Chwilowa niedostępność bazy danych czy zewnętrznego API może zniknąć po kilku sekundach, dlatego warto stosować mechanizm retry. Pamiętaj o tym, że liczba prób ma być ograniczona. Jeśli po ich osiągnięciu dalej występuje błąd, to powinno się skorzystać z Dead Letter Queue, gdzie później będzie można ją przeanalizować.

Duplikaty wiadomości i idempotencja

W wielu systemach kolejkowych należy zakładać, że ta sama wiadomość może zostać dostarczona więcej niż raz. Może się tak zdarzyć na przykład wtedy, gdy konsument poprawnie wykona operację, ale nie zdąży wysłać potwierdzenia ACK. Dobrym rozwiązaniem jest nadawanie wiadomościom unikalnego eventId i zapisywanie informacji o już obsłużonych zdarzeniach.

Monitoring

Kolejka jest dodatkowym elementem infrastruktury, dlatego powinna być monitorowana tak samo jak baza danych czy aplikacja. Dobrym rozwiązaniem jest również wykorzystanie identyfikatora wiadomości lub correlationId, który pozwala prześledzić przepływ konkretnego zdarzenia pomiędzy mikroserwisami.

Wersjonowanie wiadomości

W systemie mikroserwisowym producent i konsument mogą być wdrażani niezależnie. Zmiana struktury wiadomości po stronie producenta może więc spowodować problemy w starszej wersji konsumenta.

Warto projektować komunikaty w sposób kompatybilny wstecznie, unikać bez potrzeby usuwania lub zmiany znaczenia istniejących pól oraz w razie większych zmian stosować wersjonowanie zdarzeń.

Podsumowanie

Kolejki są jednym z podstawowych mechanizmów wykorzystywanych do przetwarzania zadań. Mogą występować lokalnie (np. BlockingQueue) albo jako zewnętrzne brokery: on premise (np. ActiveMQ) lub w chmurze (np. SNS dla AWS).

Największą zaletą kolejek jest możliwość rozdzielenia producenta od konsumenta. Poszczególne elementy systemu mogą działać niezależnie, skalować się osobno i lepiej radzić sobie z chwilową niedostępnością innych usług. Trzeba jednak pamiętać, że komunikacja oparta na wiadomościach wprowadza dodatkową złożoność, między innymi retry, duplikaty wiadomości czy konieczność monitorowania.