Jak ładować wiadomości w RabbitMQ

Question

Po otrzymaniu wiadomości, konsument/pracownik dokonuje pewnych weryfikacji, a następnie wywołuje usługę WWW. W tej fazie, jeśli wystąpi jakikolwiek błąd lub sprawdzanie poprawności nie powiedzie się, chcemy, aby wiadomość została przywrócona do kolejki, z której została oryginalnie pobrana.

Przeczytałem dokumentację RabbitMQ. Ale jestem zdezorientowany różnicami między metodami odrzucania, nack i anulowania.

Answer 1

Krótka odpowiedź:

Aby ponownie kolejkuje konkretny komunikat można odebrać zarówno basic.reject lub basic.nack z multiple flaga ustawiona na false.

basic.consume Wywołanie może również spowodować ponowne dostarczanie wiadomości, jeśli używasz potwierdzenia wiadomości i nie są potwierdzane wiadomości na konsumenta w określonym czasie i wyjścia klienta, bez ich potwierdzania.

basic.recover spowoduje ponowne dostarczenie wszystkich nie-potwierdzonych wiadomości na określonym kanale.

Długa odpowiedź:

basic.reject i basic.nack zarówno służy do tego samego celu - spadku lub ponownie kolejkuje wiadomości, które nie mogą być obsługiwane przez konkretnego konsumenta (w danym momencie, w określonych warunkach lub w ogóle). Główną różnicą między nimi jest to, że basic.nack obsługuje przetwarzanie wiadomości masowych, podczas gdy basic.reject nie.

Różnica ta opisana w Negative Acknowledgements artykule na oficjalnej stronie RabbitMQ internetowej:

Specyfikacja AMQP definiuje metodę basic.reject, który pozwala klientom odrzucić indywidualna, wydana wiadomości, instruowania brokera albo odrzucić je lub ponownie kolejkuje je. Niestety, basic.reject nie zapewnia wsparcia dla zbiorczego negatywnego potwierdzania wiadomości.

Aby rozwiązać ten problem, RabbitMQ obsługuje metodę basic.nack który zapewnia pełną funkcjonalność basic.reject podczas również pozwalając na przetwarzanie luzem wiadomości.

Aby odrzucić wiadomości masowo, klienci ustawiają flagę multiple metody basic.nack na true. Następnie broker odrzuci wszystkie niepotwierdzone, dostarczone wiadomości do wiadomości włącznie z wiadomością określoną w polu delivery_tag metody . Pod tym względem basic.nack uzupełnia semantykę potwierdzenia zbiorczego basic.ack.

Zauważ, że metoda jest RabbitMQ basic.nack specyficzne rozszerzenie podczas basic.reject sposób jest częścią AMQP 0.9.1 opisie.

Jeśli chodzi o metodę basic.cancel, powiadomiono serwer, że klient przestaje otrzymywać komunikaty. Zauważ, że klient może otrzymać dowolny numer wiadomości między metodą basic.cancel wysyłając otrzymanie odpowiedzi cancel-ok. Jeśli potwierdzenie wiadomości jest używane przez klienta i ma ona nierozpoznane komunikaty, zostaną one przeniesione z powrotem do kolejki, z której zostały pierwotnie pobrane.

basic.recover ma pewne ograniczenia w RabbitMQ: to - basic.recover with requeue=false - basic.recover synchronicity

Oprócz errata, according to RabbitMQ specsbasic.recover ma częściowe wsparcie

uwaga na temat (odzysk z ponownie kolejkuje = false nie jest obsługiwany). basic.consume:

Gdy basic.consume rozpoczęło się bez automatycznego potwierdzenia (no­ack=false) i jest kilka oczekujących komunikatów, które nie zostały potwierdzone, a następnie, gdy konsument zostanie anulowany (zginie, błąd krytyczny, wyjątek, cokolwiek), oczekujące wiadomości zostaną ponownie dostarczone. Technicznie rzecz biorąc, oczekujące wiadomości nie będą przetwarzane (nawet w formie wytłoczonej), dopóki użytkownik ich nie zwolni (ack/nack/reject/recover). Dopiero potem zostaną przetworzone (np. Martwe).

Na przykład, powiedzmy zamieścimy pierwotnie 5 wiadomości w jednym rzędzie:

Queue(main) (tail) { [4] [3] [2] [1] [0] } (head) 

A potem zużywają 3 z nich, ale nie ack je, a następnie anulować konsumenta. Będziemy mieć taką sytuację:

Queue(main) (tail) { [4] [3] [2*] [1*] [0*] } (head) 

gdzie gwiazdka (*) zauważa, że ​​redelivered flaga ustawiona na true.

Załóżmy, że mamy sytuację z martwych-literami zestawu wymiany i kolejki na śmierć zgłoskami wiadomości

Exchange(e-main)         Exchange(e-dead) 
    Queue(main){x-dead-letter-exchange: "e-dead"}  Queue(dead) 

i zakładamy zamieścimy 5 wiadomość z expire nieruchomości ustawiony 5000 (5 sek):

Queue(main) (tail) { [4] [3] [2] [1] [0] } (head) 
Queue(dead) (tail) { }(head) 

i spożywać 3 Wiadomo z main kolejki i utrzymać je na 10 sekund:

Queue(main) (tail) { [2!] [1!] [0!] } (head) 
Queue(dead) (tail) { [4*] [3*] } (head) 

gdzie wykrzyknik (!) oznacza nieprzyjęty komunikat. Takie wiadomości nie mogą być dostarczane do żadnego konsumenta i zwykle nie mogą być wyświetlane w panelu zarządzania. Ale bądźmy anulować konsumenta, należy pamiętać, że nadal posiadają 3 un-potwierdzony wiadomość:

Queue(main) (tail) { } (head) 
Queue(dead) (tail) { [2*] [1*] [0*] [4*] [3*] } (head) 

Więc teraz, że 3 wiadomości, które było w głowie umieścić z powrotem do pierwotnego kolejce, ale ponieważ posiada zestaw per-wiadomości TTL, są martwymi literami do końca kolejki listów martwych (pewnie przez wymianę listów martwych).

P.S .:

Spożywanie wiadomość aka słuchając nowego jest w jakiś sposób różni się od bezpośredniego dostępu do kolejki (coraz jedną lub więcej wiadomości bez dbanie o innych). Więcej informacji można znaleźć w opisie metody basic.get.