Kolejka (kolejka) — struktura danych na dysku lub w pamięci operacyjnej, która przechowuje odnośniki do wiadomości i udostępnia ich kopie consumers (konsumentom). Kolejka to ze stanem (gdzie mogą być buforowane same wiadomości). 1 tysiąc kolejek może zajmować około 80Mb.
Binding (wiązanie) — reguła, która informuje wymiennik, do której z kolejek mają trafiać wiadomości.
Spis treści
- RabbitMQ. Część 4. Zrozumienie, czym są wiadomości i ramki
- RabbitMQ. Część 5. Wydajność publikacji i konsumpcji wiadomości
- RabbitMQ. Część 6. Przegląd modułów Federation i Shovel
- RabbitMQ. Część 7. Szczegółowe informacje na temat połączenia i kanału
- RabbitMQ. Część 8. RabbitMQ w .NET
- RabbitMQ. Część 9. Monitorowanie
Kolejki tymczasowe
Jeśli tworzenie kolejki jest realizowane z ustawionym parametrem autoDelete, to taka kolejka zyskuje zdolność automatycznego usuwania się. Takie kolejki zwykle są tworzone w momencie podłączenia pierwszego klienta i usuwane w momencie, gdy wszyscy klienci się rozłączają.
Jeśli tworzenie kolejki jest realizowane z ustawionym parametrem ekskluzywna, to taka kolejka pozwala na podłączenie tylko jednego konsumenta i jest usuwana, gdy kanał zostanie zamknięty. Dopóki kanał nie zostanie zamknięty, klient może się rozłączać/podłączać, ale tylko w ramach tego samego połączenia. Jeśli parametr ekskluzywna jest ustawiony, to parametr autoDelete nie ma żadnego efektu.
Cechy:
- przy krótkotrwałym przerwaniu łączności stracimy wiadomości, które jeszcze nie dotarły do konsumenta
- można uchwycić zjawisko
binding churn. Zjawisko występuje, gdy liczba operacji tworzenia/usuwania kolejek i wiązań osiąga bardzo duże wartości. W trybie klastrowym taki strumień operacji będzie się rozprzestrzeniał po wszystkich węzłach i stworzy duże obciążenie. Ten proces można zoptymalizować poprzez kontrolowanie liczby subskrypcji
Stałe kolejki
Jeśli tworzenie kolejki jest realizowane z ustawionym parametrem durable, to taka kolejka zachowują swój stan i przywracają się po ponownym uruchomieniu serwera/brokera. Taka kolejka będzie istniała, dopóki nie zostanie wywołana komenda Queue.Delete.
Kolejki wysoko dostępne
Kolejki HA wymagają klastra RabbitMQ. W trybie klastrowym wszystkie informacje o wymiennikach, kolejkach, wiązaniach i konsumentach będą kopiowane na wszystkie węzły.
Gdy wiadomość jest publikowana w jakiejś kolejce HA, jest przechowywana na każdym węźle, który należy do kolejki HA. Po tym, jak wiadomość jest konsumowana na którymś z węzłów, wszystkie kopie tej wiadomości są usuwane na innych węzłach.
Kolejki HA mogą obejmować wszystkie węzły w danym klastrze lub tylko poszczególne.
Cechy:
- używanie kolejek HA prowadzi do kar w wydajności. Przy umieszczaniu wiadomości w jakiejś kolejce HA lub przy pobieraniu wiadomości z kolejki HA, RabbitMQ musi koordynować wszystkie węzły (2-3 węzły zazwyczaj wystarczą)
Tworzenie kolejki
Tworzenie kolejki odbywa się za pomocą synchronizacji RPC żądania do serwera. Żądanie jest realizowane przy użyciu metody Queue.Declare, wywoływanej z parametrami:
- nazwa kolejki
- inne parametry
Przykład tworzenia kolejki za pomocą :
// ...
channel.QueueDeclare(
queue: "my_queue",
durable: false,
exclusive: false,
autoDelete: false,
arguments: null
);
// ...queue— nazwa kolejki, którą chcemy stworzyć. Nazwa musi być unikalna i nie może być taka sama jak systemowa nazwa kolejkidurable— jeśli true, to kolejka będzie zachowywać swoje state i będzie przywracana po ponownym uruchomieniu serwera/brokeraekskluzywna— jeśli true, to kolejka będzie pozwalać na podłączenie tylko jednego konsumentaautoDelete— jeśli true, to kolejka uzyskuje zdolność automatycznego usuwania sięarguments— opcjonalne argumenty. Poniżej omówimy to bardziej szczegółowo.
arguments
x-message-ttl(x-message-time-to-live) — pozwala ustawić czas życia wiadomości w milisekundach. Jeśli tworzenie kolejki odbywa się z ustawioną wartością argumentux-message-ttl, to taka kolejka będzie automatycznie usuwać wiadomości, których czas ważności upłynął. Ustawienie wartości argumentux-message-ttlokreśla maksymalny wiek dla wszystkich wiadomości w tej kolejce. Tworzenie takiej kolejki umożliwia zapobieganie otrzymywaniu przestarzałych informacji. Można to wykorzystać w systemach rzeczywistych. Jeśli dla kolejki, która ma wymiennik do odrzuconych wiadomości, ustawić wartość argumentux-message-ttl, to odrzucone wiadomości w tej kolejce zaczną mieć czas życia.x-expires— ustawia wartość w milisekundach, po której zachodzi usunięcie kolejki. Kolejka może wydać swój czas życia tylko wtedy, gdy nie ma żadnych subskrybentów. Jeśli do kolejki są podłączeni subskrybenci, może ona automatycznie usunąć się tylko wtedy, gdy wszyscy subskrybenci wywołająBasic.Cancellub się odłączą. Czas życia kolejki może się zakończyć tylko w przypadku, gdy nie było do niej zapytaniaBasic.Get. W przeciwnym razie obecna wartość ustawienia czasu życia zostaje zresetowana, a kolejka nie będzie już automatycznie usuwana. Również nie ma gwarancji, jak szybko nastąpi usunięcie kolejki po upływie jej czasu życia.x-max-length— ustala maksymalną liczbę wiadomości w kolejce. Gdy liczba wiadomości w kolejce zacznie przekraczać maksymalną ilość, usuwane będą najstarsze.
x-max-length-bytes— ustala maksymalny dozwolony sumaryczny rozmiar ładunku wiadomości w kolejce. Po przekroczeniu ustalonej wartości (wystąpiło przepełnienie kolejki podczas publikacji wiadomości), najstarsze wiadomości zaczną być usuwane.x-overflow— ten argument służy do konfiguracji zachowania w przypadku przepełnienia kolejki. Dostępne są dwie wartości:drop-head(wartość domyślna) orazreject-publish. Jeśli wybierzeszdrop-head, najstarsze wiadomości będą usuwane. Jeśli wybierzeszreject-publish, przyjmowanie wiadomości zostanie wstrzymane.x-dead-letter-exchange— ustala exchange, do którego kierowane są odrzucone wiadomości, które nie zostały ponownie włożone do kolejki.x-dead-letter-routing-key— ustala opcjonalny klucz routingu dla odrzuconych wiadomości.x-max-priority— umożliwia sortowanie według priorytetów w kolejce z maksymalną wartością priorytetu 255 (RabbitMQ wersje 3.5.0 i wyższe). Liczba wskazuje maksymalny priorytet, który będzie wspierany przez kolejkę. Jeśli argument nie jest ustawiony, kolejka nie będzie wspierać priorytetów wiadomości.x-queue-mode— umożliwia przełączenie kolejki w tryb leniwy. W tym trybie jak najwięcej wiadomości będzie przechowywanych na dysku. Użycie pamięci operacyjnej będzie minimalne. W przypadku, gdy nie jest ustawiony, kolejka będzie przechowywać wiadomości w pamięci, aby dostarczać wiadomości jak najszybciej.x-queue-master-locator— jeśli mamy klaster, można ustawić master queue.x-ha-policy— jest używany podczas tworzenia HA kolejek i określa, jak wiadomość będzie rozpowszechniana na węzłach. Jeśli ustawiona jest wartośćwszystko, wiadomość będzie przechowywana na wszystkich węzłach. Jeśli ustawiona jest wartośćnodes, wiadomość będzie przechowywana na określonych węzłach klastra.x-ha-nodes— ustala węzły, do których będzie należała dana kolejka.HA
Jeśli utworzenie kolejki być może, to serwer wyśle do klienta synchronizowane RPC odpowiedzi Queue.DeclareOk. Jeśli utworzenie kolejki niemożliwe (wystąpił błąd w żądaniu Queue.Declare), to kanał zostanie zamknięty serwerem za pomocą polecenia Channel.Close i klient otrzyma wyjątek , który będzie zawierał kod błędu i jego opis.
Ponowne wywołanie Queue.Declare z tymi samymi parametrami zwróci użyteczne informacje o tej kolejce. Na przykład, ogólną liczbę wiadomości oczekujących w danej kolejce oraz ogólną liczbę subskrybentów na nią.
Wywołanie Queue.Declare z danymi logowania użytkownika, któremu nie przypisano wymaganych uprawnień zamknie kanał za pomocą polecenia Channel.Close i klient otrzyma wyjątek , który będzie zawierał kod błędu i jego opis.
Po tym, jak kolejka jest bezczynna przez >= 10 sekund, wchodzi w stan uśpienia, wywołując GC w kolejce, co prowadzi do znacznego zmniejszenia pamięci potrzebnej dla tej kolejki.
Tworzenie kolejki za pomocą interfejsu graficznego
Zaloguj się do panelu administracyjnego RabbitMQ jako użytkownik gość (nazwa użytkownika: gość i hasło: gość). Zauważ, że użytkownik gość może łączyć się tylko z lokalnego hosta. Teraz przejdź do zakładki Kolejki i kliknij na Dodaj nową kolejkę. Wypełnij właściwości:
Po wprowadzeniu wszystkich wymaganych danych i naciśnięciu na Dodaj kolejki, kolejka pojawi się na ogólnej liście.
Kliknięcie na nazwę kolejki pokaże jej szczegółowe informacje. Tutaj można skonfigurować powiązanie między wymianą a kolejką, zobaczyć listę consumers, publikować/odbierać wiadomości, usunąć kolejkę i zobaczyć statystyki.
Tworzenie powiązania
Tworzenie powiązania odbywa się za pomocą synchronizacji RPC żądania do serwera. Żądanie jest realizowane przy użyciu metody Queue.Bind, wywoływanej z parametrami:
- nazwa kolejki
- nazwa punktu wymiany
- inne parametry
Przykład tworzenia powiązania za pomocą :
//...
channel.QueueBind(
queue: queueName,
exchange: "my_exchange",
routingKey: "my_key",
arguments: null
);
//...queue— nazwa kolejkiexchange— nazwa wymiennikaroutingKey— klucz routinguarguments— opcjonalne argumenty
Jeśli tworzenie powiązania być może, to serwer wyśle do klienta synchronizowane RPC odpowiedzi Queue.BindOk.
Tworzenie powiązania za pomocą interfejsu graficznego
Zaloguj się do panelu administracyjnego RabbitMQ jako użytkownik gość (nazwa użytkownika: gość i hasło: gość). Zauważ, że użytkownik gość może łączyć się tylko z lokalnego hosta. Teraz przejdź do zakładki Kolejki i klikamy na kolejkę my_queue. Wypełniamy pola sekcji bindings:
Po wprowadzeniu wszystkich wymaganych danych i naciśnięciu na Powiąż, powiązanie pojawi się na ogólnej liście:
Kod
W tej sekcji opiszemy kolejkę i powiązanie z kodem w C#, tak jakbyśmy musieli stworzyć bibliotekę. Może to być przydatne dla zrozumienia.
public interface IQueue
{
string Name { get; }
//
// Jeśli ustawić na true, kolejka będzie trwała.
// Będzie przechowywana na dysku i może
// przetrwać restart serwera/brokera.
// Jeśli wartość to false, kolejka jest tymczasowa i będzie usuwana,
// gdy serwer/broker zostanie ponownie uruchomiony
//
bool IsDurable { get; }
//
// Jeśli wartość to true,
// ta kolejka będzie pozwalać na podłączenie
// tylko jednemu consumer-owi
//
bool IsExclusive { get; }
//
// Automatyczne usunięcie.
// Kolejka zostanie usunięta, gdy wszyscy klienci się odłączą.
//
bool IsAutoDelete { get; }
//
// Opcjonalne argumenty
//
IDictionary Arguments { get; }
}public class Queue : IQueue
{
public Queue(
string name,
bool isDurable = true,
bool isExclusive = false,
bool isAutoDelete = false,
IDictionary arguments = null)
{
Name = name ??
throw new ArgumentNullException(name, $"{name} must not be null");
IsDurable = isDurable;
IsExclusive = isExclusive;
IsAutoDelete = isAutoDelete;
Arguments = arguments ?? new Dictionary();
}
public string Name { get; }
public bool IsDurable { get; }
public bool IsExclusive { get; }
public bool IsAutoDelete { get; }
public IDictionary Arguments { get; }
}public static class QueueMode
{
public const string Default = "default";
//
// Tryb leniwy. Tryb leniwy zmusi do zachowania
// jak największej liczby wiadomości na dysku, aby zminimalizować
// użycie pamięci RAM
//
public const string Lazy = "lazy";
}public interface IBinding
{
//
// Wymiennik, który będzie wiązany z powiązaniem
//
IExchange Exchange { get; }
//
// Klucz routingu
//
string RoutingKey { get; }
//
// Opcjonalne argumenty
//
IDictionary Arguments { get; }
}public class Binding : IBinding
{
public Binding(
IExchange exchange,
string routingKey,
IDictionary arguments)
{
Exchange = exchange;
RoutingKey = routingKey;
Arguments = arguments;
}
public IExchange Exchange { get; }
public string RoutingKey { get; }
public IDictionary Arguments { get; }
}Źródło: habr.com
