🥇RabbitMQ. 3. rész. Fedezzük fel a sorokat (Queues) és a kötéseket (Bindings)

Sor (sor) — olyan adattípus a lemezen vagy a memóriában, amely hivatkozásokat tárol az üzenetekre, és visszaadja azok másolatait consumers (fogyasztóknak). Sor jelentése Erlang-folyamat állapottal (ahol maguk az üzenetek is cache-elhetők). 1 ezer sor körülbelül 80 MB-ot foglalhat.

Binding (kötés) — egy szabály, amely megmondja a cserélőnek, hogy melyik sorba kell az üzeneteknek kerülniük.

Tartalomjegyzék

RabbitMQ. 1. rész. Bevezetés. Erlang, AMQP és RPC
RabbitMQ. 2. rész. Az Exchanges részletezése
RabbitMQ. 3. rész. Queues és Bindings megértése
RabbitMQ. 4. rész. Nézzük meg, hogy mik azok az üzenetek és keretek
RabbitMQ. 5. rész. Az üzenetek közzétételi és fogyasztási teljesítménye
RabbitMQ. 6. rész. A Federation és Shovel modulok áttekintése
RabbitMQ. 7. rész. Részletesen a Kapcsolatról és csatornáról
RabbitMQ. 8. rész. RabbitMQ .NET környezetben
RabbitMQ. 9. rész. Monitoring

Ideiglenes sorok

Ha a sor létrehozása egy megadott paraméterrel történik autoDelete, akkor ez a sor olyan képességet kap önállóan eltávolítja magát. Ezek a sorok általában akkor jönnek létre, amikor az első kliens csatlakozik, és akkor törlődnek, amikor minden kliens lecsatlakozik.

Ha a sor létrehozása egy megadott paraméterrel történik kizárólagos, akkor ez a sor csak egy fogyasztónak engedélyezi a csatlakozást és eltávolítódik, ha a csatorna bezáródik. Addig, amíg a csatorna nyitva van, a kliens el tudja távolítani/csatlakozni, de csak ugyanazon a kapcsolaton belül. Ha a paraméter kizárólagos be van állítva, akkor a autoDelete paraméternek nincs hatása.

Jellemzők:

rövid idejű kapcsolatzavar esetén elveszítjük azokat az üzeneteket, amelyek még nem jutottak el a fogyasztóhoz
lehet elkapni a jelenséget binding churn. A jelenség akkor fordul elő, amikor a sorok és kötéseik létrehozására/törlésére vonatkozó műveletek száma nagyon magas értéket ér el. Klaszter módban ez a műveletsorozat az összes csomóponton szét fog terjedni, és nagy terhelést fog létrehozni. Ezt a folyamatot optimalizálni lehet a feliratkozások számának ellenőrzésével

Állandó sorok

Ha a sor létrehozása egy megadott paraméterrel történik durable, akkor ez a sor megőrzi az állapotát és helyreáll a szerver/broker újraindítása után. Ez a sor addig fog létezni, amíg a Queue.Delete.

parancsot nem hívják meg.

Magas rendelkezésre állású sorok

A HA sorok klaszter környezetet igényelnek a RabbitMQ számára. Klaszter módban az összes információ a cserélőkről, sorokról, kötésekről és fogyasztókról másolódik az összes csomópontba.

Amikor egy üzenet publikálva van egy HA sorba, akkor az minden csomóponton tárolódik, amely a HA sorhoz tartozik. Miután egy üzenetet egy csomóponton fogyasztanak, az üzenet minden más csomópontban lévő másolata törlésre kerül.

Jellemzők:

A HA queue's use can lead to performance penalties. When placing a message into a HA queue or consuming a message from it, RabbitMQ must perform coordination across all nodes (2-3 nodes are usually sufficient).

Queue létrehozása

A queue létrehozása szinkron RPC lekéréssel a szerverhez. A kérés a következő metódussal történik: Queue.Declare, az alábbi paraméterekkel:

a queue neve
egyéb paraméterek

Példa queue létrehozására RabbitMQ.Client:

// ...
channel.QueueDeclare(
    queue: "my_queue",
    durable: false,
    exclusive: false,
    autoDelete: false,
    arguments: null
);
// ...

queue — a queue neve, amelyet létre szeretnénk hozni. A névnek egyedinek kell lennie, és nem lehet azonos a rendszer queue nevével.
durable — ha igaz, a queue megőrzi az állapotát és visszaáll a szerver/broker újraindulása után. — ha igaz, a queue lehetővé teszi, hogy csak egy fogyasztó csatlakozzon.
kizárólagos — ha igaz, a queue rendelkezik azzal a képességgel,
autoDelete arguments önállóan eltávolítja magát
— választható argumentumok. Részletesebben lásd alább. x-message-ttl

— választható argumentumok. Részletesebben lásd alább.

x-message-time-to-live() — lehetővé teszi a message élettartamának beállítását milliszekundumban. Ha a queue létrehozása egy megadott értékkel történik,, akkor a queue automatikusan ki fogja zárni azokat a message-eket, amelyek múlt időtartamúak. x-message-time-to-live. Az argumentum értékének beállítása meghatározza az összes message maximális életkorát a queue-ban. Ilyen queue létrehozásamegelőzheti az elavult információk megszerzését. x-message-time-to-live . Ezt valós idejű rendszerekben lehet használni. Ha a queue-hoz, amelyhez elutasított message-ek cseréje van megadva, argumentum értéket adunk meg, , akkor az elutasított message-ek ebben a queue-banéletidővel fognak rendelkezni. x-message-time-to-livex-expires — meghatároz egy milliszekundumban kifejezett értéket, amely után a queue törlődik. A queue csak akkor veszítheti el az érvényességét, ha nincs előfizetője. Ha a queue előfizetőkkel rendelkezik, automatikusan csak akkor törlődik, amikor az összes előfizető hívja a.
Basic.Cancel vagy leválik. A queue élettartama csak úgy érhet véget, ha nem történt kérés a felé Basic.Get . Ellenkező esetben a jelenlegi élettartam-képesség visszaáll, és a queue már nem törlődik automatikusan. Továbbá Basic.Get.Otherwise, the current lifetime setting is reset, and the queue will no longer be automatically deleted. Also nincs garancia arra, hogy milyen gyorsan távolítják el a sort a lejárati ideje után.
x-max-length — meghatározza a sorban lévő üzenetek maximális számát. Ha az üzenetek száma meghaladja a maximális értéket, a legrégebbi üzenetek kezdik eltávolítani

x-max-length-bytes — meghatározza az üzenetek hasznos terhelésének maximálisan megengedett összesített méretét a sorban. Ha ezt a beállított értéket túllépik (túlcsordulás lép fel a következő üzenet közzétételekor), a legrégebbi üzenetek kezdenek eltávolítani
x-overflow — ez az argumentum a sor túlcsordulásának következményeinek beállítására szolgál. Két érték érhető el: drop-head (alapértelmezett érték) és reject-publish. Ha a drop-headopciót választjuk, a legrégebbi üzeneteket eltávolítják. Ha reject-publishopciót választjuk, az üzenetek fogadása felfüggesztésre kerül
x-dead-letter-exchange — meghatározza azt az exchange-t, ahová a visszautasított üzenetek mennek, amelyek nem kerülnek újra a sorba
x-dead-letter-routing-key — meghatározza a nem kötelező irányítási kulcsot a visszautasított üzenetekhez
x-max-priority — lehetővé teszi a sorban a prioritások szerinti rendezést a maximális prioritás 255 értékkel (RabbitMQ 3.5.0 és újabb verziók). A szám a maximális prioritást jelzi, amelyet a sor támogatni fog. Ha a paraméter nincs beállítva, a sor nem fogja támogatni az üzenetek prioritását
x-queue-mode — lehetővé teszi a sor váltását lusta módba. Ebben a módban annyi üzenetet tárolnak, amennyit csak lehet a lemezen. A memória használata minimális lesz. Ha ez nincs beállítva, a sor az üzeneteket a memóriában tárolja, hogy az üzeneteket a lehető leggyorsabban szállíthassa
x-queue-master-locator — ha van klaszterünk, be lehet állítani a mester sort
x-ha-policy — HA sorok létrehozásakor használják, és meghatározza, hogyan fog az üzenet eloszlani a csomópontok között. Ha be van állítva az érték mind, az üzenet mentésre kerül minden csomóponton. Ha az érték nodes, az üzenet csak a klaszter meghatározott csomópontjain marad meg
x-ha-nodes — meghatározza azokat a csomópontokat, amelyekhez egy sor tartozni fog HA

Ha a sor létrehozása lehetséges, a szerver szinkronikus RPC választ fog küldeni az ügyfélnek Queue.DeclareOk. Ha a sor létrehozása lehetetlen (kérelmezés során hiba történt Queue.Declare), akkor a csatorna le fog zárulni a szerver által a Channel.Close parancs segítségével és az ügyfél kivételt kap OperationInterruptedException, amely tartalmazni fog egy hiba kódot és annak leírását.

Újrahívás Queue.Declare hasonló paraméterekkel hasznos információt ad vissza erről a sorban. Például a sorban várakozó üzenetek összesített száma és a rá feliratkozott fogyasztók összesített száma.

Hívás Queue.Declare a felhasználó azonosítóval, akinek nincsenek a szükséges jogai lezárja a csatornát a parancs segítségével Channel.Close parancs segítségével és az ügyfél kivételt kap OperationInterruptedException, amely tartalmazni fog egy hiba kódot 403 és annak leírását.

Miután a sor legalább 10 másodpercig inaktív, álomba merül, GC-t hív elő a sorban, ami jelentős memória megtakarítást eredményez ennek a sorának.

Queue létrehozása a grafikus felületen

Belépünk az adminisztrátori panelre RabbitMQ a felhasználóval guest (felhasználónév: guest és jelszó: guest). Kérjük, vegye figyelembe, hogy a felhasználó guest csak a helyi gépről csatlakozhat. Most térjünk át a Sorok fülre és kattintsunk a Új sor hozzáadása. Töltse ki a tulajdonságokat:

Amint megadjuk az összes szükséges adatot és rákattintunk a Sorok hozzáadása, a sor megjelenik a főlistában.

A sor nevére kattintva megjelenik a részletes információ róla. Itt beállíthatjuk a kötődést a csere és a sor között, megtekinthetjük a listát consumers, üzeneteket publikálhatunk/megkaphatunk, törölhetjük a sort és megtekinthetjük a statisztikát.

Kötés létrehozása

A kötés létrehozása szinkron RPC lekéréssel a szerverhez. A kérés a következő metódussal történik: Queue.Bind, az alábbi paraméterekkel:

a queue neve
a csere nevét
egyéb paraméterek

Példa kötés létrehozására a RabbitMQ.Client:

//...
channel.QueueBind(
    queue: queueName,
    exchange: "my_exchange",
    routingKey: "my_key",
    arguments: null
);
//...

queue — a sor neve
exchange — a csere neve
routingKey — a routing kulcs
— választható argumentumok. Részletesebben lásd alább. — opcionális argumentumok

Ha a kötés létrehozása lehetséges, a szerver szinkronikus RPC választ fog küldeni az ügyfélnek Queue.BindOk.

Kötés létrehozása a grafikus felületen

Belépünk az adminisztrátori panelre RabbitMQ a felhasználóval guest (felhasználónév: guest és jelszó: guest). Kérjük, vegye figyelembe, hogy a felhasználó guest csak a helyi gépről csatlakozhat. Most térjünk át a Sorok és kattintsunk a sorra my_queue. Töltse ki a mezőket a szakaszban kötések:

Amint megadjuk az összes szükséges adatot és rákattintunk a Köti, a kötés megjelenik a főlistában:

Code

Ebben a részben a C# kód segítségével leírjuk a sort és a kötések, mintha könyvtárat kellene fejlesztenünk. Lehet, hogy ez hasznos lesz a megértéshez.

public interface IQueue
    {        
        string Name { get; }

        // 
        //     Ha a true értékre állítja, akkor a queue állandó lesz. 
        //     Ez a lemezen tárolódik, és túlélheti a szerver/broker újraindítását. 
        //     Ha a false értékre állítja, akkor a queue ideiglenes, és törölve lesz, 
        //     amikor a szerver/broker újraindul
        // 
        bool IsDurable { get; }

        // 
        //     Ha az érték true, akkor 
        //     ez a queue csak egy consumer-hez való kapcsolódást engedélyez
        // 
        bool IsExclusive { get; }

        // 
        //     Automatikus törlés. 
        //     A queue törlésre kerül, amikor az összes kliens leválik.
        // 
        bool IsAutoDelete { get; }

        // 
        //     Opcionális argumentumok
        // 
        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} nem lehet 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";
        // 
        //     Lusta mód. A lusta mód megköveteli, hogy a lehető legtöbb üzenetet 
        //     mentse a lemezre, hogy csökkentse a 
        //     memóri használatát
        // 
        public const string Lazy = "lazy";
    }

public interface IBinding
    {
        // 
        //     Közvetítő, amely a kötődéssel fog összekapcsolódni
        // 
        IExchange Exchange { get; }

        // 
        //     Útvonal kulcs
        // 
        string RoutingKey { get; }

        // 
        //     Opcionális argumentumok
        // 
        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; }
    }

Forrás: habr.com