🥇RabbitMQ. Časť 3. Pozrime sa na Queues a Bindings

Fronta (fronta) — dátová štruktúra na disku alebo v pamäti, ktorá uchováva odkazy na správy a poskytuje ich kópie consumers (spotrebiteľom). Fronta predstavuje Erlang proces so stavom (kde môžu byť vyrovnávané aj samotné správy). 1 tisíc front môže zaberajú približne 80Mb.

Binding (viazanie) — pravidlo, ktoré informuje výmenník, do ktorej fronty by mali správy ísť.

Obsah

RabbitMQ. Časť 1. Úvod. Erlang, AMQP a RPC
RabbitMQ. Časť 2. Rozoberieme si Exchanges
RabbitMQ. Časť 3. Pochopenie Queues a Bindings
RabbitMQ. Časť 4. Pochopíme, čo sú správy a rámce
RabbitMQ. Časť 5. Výkon publikovania a spotreby správ
RabbitMQ. Časť 6. Prehľad modulov Federation a Shovel
RabbitMQ. Časť 7. Podrobne o Connection a Channel
RabbitMQ. Časť 8. RabbitMQ v .NET
RabbitMQ. Časť 9. Monitorovanie

Dočasné fronty

Ak sa fronta vytvorí s nastaveným parametrom autoDelete, potom taká fronta získava schopnosť automaticky sa odstraňovať. Tieto fronty sa zvyčajne vytvárajú v momente pripojenia prvého klienta a odstraňujú sa, keď sa všetci klienti odpoja.

Ak sa fronta vytvorí s nastaveným parametrom exkluzívne, potom taká fronta umožňuje pripojiť sa iba jednému spotrebiteľovi a odstráni sa, ak sa kanál uzavrie. Pokiaľ kanál nie je uzavretý, klient sa môže odpojiť/pripojiť, ale iba v rámci toho istého pripojenia. Ak je parameter exkluzívne nastavený, potom parameter autoDelete nemá žiadny účinok.

Funkcie:

pri krátkodobom prerušení spojenia stratíme správy, ktoré ešte nestihli doraziť k spotrebiteľovi
môžeme zažiť fenomén binding churn. Fenomén vzniká, keď počet operácií vytvárania/odstraňovania front a viazaní dosiahne veľmi veľké hodnoty. V klastrovom režime sa takýto tok operácií rozšíri po všetkých uzloch a spôsobí veľké zaťaženie. Tento proces je možné optimalizovať pomocou kontroly počtu predplatných.

Trvalé fronty

Ak sa fronta vytvorí s nastaveným parametrom durable, potom taká fronta uchovávajú svoj stav a obnovujú sa po reštarte servera/brokera. Táto fronta bude existovať až do vyvolania príkazu Queue.Delete.

Vysoko dostupné fronty

Fronty HA vyžadujú klastrové prostredie RabbitMQ. V klastrovom režime budú všetky informácie o výmenníkoch, frontách, viazaniach a spotrebiteľoch skopírované na všetky uzly.

Keď je správa zverejnená do nejakej fronty HA, uchováva sa na každom uzle patriacom k frontám HA. Po tom, čo je správa spracovaná na niektorom z uzlov, všetky kópie tejto správy budú odstránené na ostatných uzloch.

Fronty HA môžu byť rozšírené na všetky uzly v určitom klastri alebo iba na jednotlivé.

Funkcie:

používanie HA front vedie k penalizáciám v oblasti výkonu. Pri umiestnení správy do nejakej HA fronty alebo pri spracovaní správy z HA fronty musí RabbitMQ vykonávať koordináciu medzi všetkými uzlami (2-3 uzly sú zvyčajne dostatočné).

Vytvorenie fronty

Vytvorenie fronty sa uskutočňuje pomocou synchronizácie. RPC žiadania na server. Žiadosť sa uskutočňuje pomocou metódy Queue.Declare, vyžaduje si parametre:

názov fronty
iné parametre

Príklad vytvorenia fronty pomocou RabbitMQ.Client:

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

queue — názov fronty, ktorú chceme vytvoriť. Názov musí byť jedinečný a nemôže sa zhodovať so systémovým názvom fronty.
durable — ak true, tak fronta bude uchovávať svoj stav a obnoví sa po reštartovaní servera/brokera.
exkluzívne — ak true, tak fronta umožní pripojiť sa len jednému spotrebiteľovi.
autoDelete — ak true, tak fronta nadobudne schopnosť automaticky sa odstraňovať
argumentoch — voliteľné argumenty. Nižšie si podrobnejšie vysvetlíme.

argumentoch

x-message-ttl(x-message-time-to-live) — umožňuje nastaviť čas do vypršania životnosti správy v milisekundách. Ak sa fronta vytvorí s nastavenou hodnotou argumentu x-message-ttl, taká fronta bude automaticky vylučovať správy, ktorým vypršal čas.Nastavenie hodnoty argumentu x-message-ttl určuje maximálny vek pre všetky správy v danej fronte. Vytvorenie takej fronty umožňuje zabrániť prijímaniu zastaraných informácií.To sa dá využiť v systémoch reálneho času. Ak má fronta, pre ktorú je nastavený výmenník pre odmietnuté správy, nastavenú hodnotu argumentu x-message-ttl, tak odmietnuté správy v danej fronte začnú mať životnosť..
x-expires — nastavuje hodnotu v milisekundách, po ktorej dôjde k odstráneniu fronty. Fronta môže vyčerpať svoju dobu platnosti iba v prípade, že nemá žiadnych odberateľov. Ak sú k fronte pripojení odberatelia, môže sa automaticky odstrániť iba vtedy, keď sa všetci odberatelia odpoja. Basic.Cancel alebo sa odpoja. Doba platnosti fronty môže skončiť iba v prípade, že k nej nebol požiadavok Basic.Get. Inak sa aktuálna hodnota nastavenia doby platnosti vynuluje a fronta sa už nebude automaticky odstraňovať. Taktiež nie sú záruky, ako rýchlo dochádza k odstráneniu fronty po uplynutí jej doby platnosti..
x-max-length — určuje maximálny počet správ v fronte. Ak počet správ vo fronte začne prekračovať maximálny limit, začnú sa odstraňovať najstaršie

x-max-lenght-bytes — určuje maximálne povolenú celkovú veľkosť payloadu správ vo fronte. Pri prekročení nastavenej hodnoty (došlo k preplneniu fronty pri ďalšom publikovaní správy) sa najstaršie správy začnú odstrániť
x-overflow — tento argument sa používa na nastavenie správania v prípade preplnenia fronty. Dostupné sú dve hodnoty: drop-head (predvolená hodnota) a reject-publish. Ak zvolíte drop-head, najstaršie správy budú odstránené. Ak zvolíte reject-publish, príjem správ bude pozastavený
x-dead-letter-exchange — určuje exchange, do ktorého sú smerované odmietnuté správy, ktoré neboli opätovne zaradené do fronty
x-dead-letter-routing-key — určuje voliteľný kľúč pre smerovanie odmietnutých správ
x-max-priority — umožňuje triedenie podľa priorít vo fronte s maximálnou hodnotou priority 255 (RabbitMQ verzií 3.5.0 a vyššej). Číslo uvádza maximálnu prioritu, ktorú fronta bude podporovať. Ak argument nie je nastavený, fronta nebude podporovať prioritu správ
x-queue-mode — umožňuje prepnúť frontu do lenivého režimu. V tomto režime bude čo najviac správ uložených na disku. Používanie operačnej pamäte bude minimálne. V prípade, že nie je nastavený, fronta bude uchovávať správy v pamäti, aby dodávala správy čo najrýchlejšie
x-queue-master-locator — ak máme klaster, môžeme nastaviť hlavný uzol fronty
x-ha-policy — používa sa pri vytváraní HA front a určuje, ako sa správa bude šíriť medzi uzlami. Ak je nastavená hodnota všetko, správa sa uloží na všetkých uzloch. Ak je nastavená hodnota nodes, správa sa uloží na konkrétnych uzloch v klastri
x-ha-nodes — určuje uzly, ku ktorým bude patriť daná fronta HA

Ak je vytváranie fronty možno, server zašle klientovi synchronné RPC odpovedi Queue.DeclareOk. Ak je vytváranie fronty nedá vykonať (nastal problém s požiadavkou Queue.Declare), potom kanál sa zatvorí serverom prostredníctvom príkazu Channel.Close a klient dostane výnimku OperationInterruptedException, ktorá bude obsahovať kód chyby a jej opis.

Opakované vyvolanie Queue.Declare s identickými parametrami vráti užitočné informácie o tejto fronte. Napríklad celkový počet správ čakajúcich v tejto fronte a celkový počet prihlásených spotrebiteľov.

Vyvolanie Queue.Declare pod používateľskými údajmi, ktorým neboli pridelené potrebné práva uzavrie kanál pomocou príkazu Channel.Close a klient dostane výnimku OperationInterruptedException, ktorý bude obsahovať chybový kód 403 a jeho popis.

Akonáhle fronta vyčkáva po dobu >= 10 sekúnd, prejde do spánkového režimu, spôsobujúc GC vo fronte, čo vedie k značnému zníženiu pamäte potrebnej pre tento front.

Vytvorenie fronty cez grafické užívateľské rozhraní

Prihláste sa do ovládacieho panela RabbitMQ pod používateľom návštevník (username: návštevník a password: návštevník). Všimnite si, že používateľ návštevník sa môže pripojiť iba z lokálneho hostiteľa. Teraz prejdeme na kartu Fronty a klikneme na Pridať novú frontu. Vyplňte vlastnosti:

Keď všetky potrebné údaje zadáme a klikneme na Pridať fronty, fronta sa objaví v celkovom zozname.

Kliknutím na názov fronty sa zobrazia jej podrobnosti. Tu môžeme nastaviť väzbu medzi výmenou a frontou, zobraziť zoznam consumers, publikovať/získať správy, vymazať frontu a zobraziť štatistiku.

Vytvorenie väzby

Vytvorenie väzby prebieha cez synchronné RPC žiadania na server. Žiadosť sa uskutočňuje pomocou metódy Queue.Bind, vyžaduje si parametre:

názov fronty
názov výmenného bodu
iné parametre

Príklad vytvorenia väzby cez RabbitMQ.Client:

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

queue — názov fronty
exchange — názov výmenníka
routingKey — kľúč smerovania
argumentoch — voliteľné argumenty

Ak vytvorenie väzby možno, server zašle klientovi synchronné RPC odpovedi Queue.BindOk.

Vytvorenie väzby cez grafické užívateľské rozhranie

Keď všetky potrebné údaje zadáme a klikneme na Pripojiť, väzba sa zobrazí v celkovom zozname:

Kód

V tejto sekcii opíšeme frontu a väzbu kódom v C#, akoby sme potrebovali vyvinúť knižnicu. Možno to bude užitočné na pochopenie.

public interface IQueue
    {        
        string Name { get; }

        // <summary>
        //     Ak nastavíte true, fronta bude trvalá. 
        //     Bude uložená na disku a môže 
        //     prežiť reštart servera/brokera. 
        //     Ak je hodnota false, fronta je dočasná a bude odstránená, 
        //     keď bude server/broker reštartovaný
        // </summary>
        bool IsDurable { get; }

        // <summary>
        //     Ak je hodnota true, 
        //     takáto fronta umožní pripojiť sa 
        //     iba jednému consumer-ovi
        // </summary>
        bool IsExclusive { get; }

        // <summary>
        //     Automatické odstránenie. 
        //     Fronta bude odstránená, keď sa odpoja všetci klienti.
        // </summary>
        bool IsAutoDelete { get; }

        // <summary>
        //     Voliteľné argumenty
        // </summary>
        IDictionary<string, object> 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} nesmie byť prázdny");

            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";
        // <summary>
        //     Lenivý režim. Lenivý režim spôsobí, že sa uložia 
        //     čo najviac správ na disku, aby sa znížila 
        //     spotreba operačnej pamäte
        // </summary>
        public const string Lazy = "lazy";
    }

public interface IBinding
    {
        // <summary>
        //     Burza, ktorá bude viazať spojenie
        // </summary>
        IExchange Exchange { get; }

        // <summary>
        //     Kľúč smerovania
        // </summary>
        string RoutingKey { get; }

        // <summary>
        //     Nepovinné argumenty
        // </summary>
        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; }
    }

Zdroj: habr.com