🥇RabbitMQ. Část 3. Pochopení front a vazeb

Fronta (fronta) — datová struktura na disku nebo v paměti, která uchovává odkazy na zprávy a poskytuje jejich kopie spotřebitelé (pro spotřebitele). Fronta představuje Erlang proces se stavem (kde mohou být zprávy také ukládány do cache). 1 000 front může zabírat přibližně 80 MB.

Binding (připojení) — pravidlo, které informuje výměník, do které fronty mají zprávy směřovat.

Obsah

RabbitMQ. Část 1. Úvod. Erlang, AMQP a RPC
RabbitMQ. Část 2. Zkoumáme Exchanges
RabbitMQ. Část 3. Zkoumáme Queues a Bindings
RabbitMQ. Část 4. Zkoumáme, co jsou zprávy a rámy
RabbitMQ. Část 5. Výkon publikace a spotřeby zpráv
RabbitMQ. Část 6. Přehled modulů Federation a Shovel
RabbitMQ. Část 7. Podrobně o Connection a Channel
RabbitMQ. Část 8. RabbitMQ v .NET
RabbitMQ. Část 9. Monitorování

Dočasné fronty

Pokud je fronta vytvořena s nastaveným parametrem autoDelete, pak taková fronta získává schopnost automaticky se odstranit. Tyto fronty se obvykle vytvářejí v okamžiku připojení prvního klienta a odstraňují se ve chvíli, kdy se všichni klienti odpojí.

Pokud je fronta vytvořena s nastaveným parametrem exkluzivní, pak taková fronta umožňuje připojení pouze jednomu spotřebiteli a bude odstraněna, pokud se kanál zavře. Dokud nebude kanál uzavřen, může se klient odpojovat/připojovat, ale pouze v rámci stejného spojení. Pokud je parametru exkluzivní n nastaven, pak parametr autoDelete nemá žádný účinek.

Vlastnosti:

při krátkodobém přerušení spojení budeme ztrácet zprávy, které se ještě nestihly dostat k spotřebiteli
můžeme zachytit fenomén binding churn. Tento fenomén nastává, když počet operací vytváření/odstraňování front a připojení dosáhne velmi vysokých hodnot. V klastrovém režimu se tento tok operací rozšíří po všech uzlech a vytvoří velké zatížení. Tento proces lze optimalizovat kontrolou počtu předplatného.

Trvalé fronty

Pokud je fronta vytvořena s nastaveným parametrem durable, pak taková fronta uchovávají svůj stav a obnovují se po restartu serveru/brokera. Tato fronta bude existovat, dokud nebude provedena objednávka Queue.Delete.

Vysoko dostupné fronty

Fronty HA vyžadují clustrové prostředí RabbitMQ. V klastrovém režimu budou všechny informace o výměnách, frontách, připojeních a spotřebitelích zkopírovány na všechny uzly.

Když je zpráva publikována do nějaké HA fronty, je uchovávána na každém uzlu, který patří k HA frontě. Jakmile je zpráva zpracována na některém z uzlů, všechny kopie této zprávy budou odstraněny na ostatních uzlech.

Fronty HA mohou být rozšířeny na všechny uzly v určitém clusteru nebo pouze na jednotlivé.

Vlastnosti:

použití HA front vede k penalizacím v výkonu. Když se zpráva umístí do určité HA fronty nebo se zpráva spotřebovává z HA fronty, RabbitMQ musí provést koordinaci mezi všemi uzly (2-3 uzly obvykle stačí)

Vytvoření fronty

Vytvoření fronty se provádí pomocí synchronního RPC žádosti k serveru. Žádost je realizována pomocí metody Queue.Declare, volané s parametry:

název fronty
další parametry

Příklad vytvoření fronty pomocí RabbitMQ.Client:

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

fronta — název fronty, kterou chceme vytvořit. Název musí být jedinečný a nesmí se shodovat se systémovým názvem fronty
durable — pokud je true, fronta bude uchovávat svůj stav a obnoví se po restartu serveru/brokera
exkluzivní — pokud je true, fronta umožní připojit se pouze jednomu spotřebiteli
autoDelete — pokud je true, fronta získává schopnost automaticky se odstranit
arguments — volitelné argumenty. Níže rozebere podrobněji.

arguments

x-message-ttl(x-message-time-to-live) — umožňuje nastavit časový limit zprávy v milisekundách. Pokud se fronta vytváří se nastavenou hodnotou argumentu x-message-ttl, pak taková fronta bude automaticky vylučovat zprávy, jejichž platnost vypršela. Nastavení hodnoty argumentu x-message-ttl určuje maximální věk všech zpráv v této frontě. Vytvoření takové fronty umožňuje zabránit příjmu zastaralých informací. To lze použít v reálných systémech. Pokud má fronta, pro kterou je nastavena výměna pro odmítnuté zprávy, nastavenou hodnotu argumentu x-message-ttl, pak odmítnuté zprávy v této frontě začnou mít dobu platnosti.
x-expires — určuje hodnotu v milisekundách, po jejímž uplynutí dojde k odstranění fronty. Fronta může vyčerpat svou životnost pouze v případě, že nemá žádné odběratele. Pokud jsou k frontě připojeni odběratelé, může se automaticky odstranit pouze tehdy, když všichni odběratelé vyvolají Basic.Cancel nebo se odpojí. Doba života fronty může vypršet pouze v případě, že k ní nebyl požadavek Basic.Get. Jinak se aktuální hodnota nastavení životnosti resetuje a fronta se již nebude automaticky odstraňovat. Také není zaručeno, jak rychle dojde k odstranění fronty po uplynutí jejího životního cyklu.
x-max-length — určuje maximální počet zpráv ve frontě. Pokud počet zpráv ve frontě začne překračovat maximální počet, začnou se odstraňovat nejstarší

x-max-length-bytes — určuje maximální povolenou celkovou velikost payloadu zpráv ve frontě. Při překročení stanovené hodnoty (dochází k přetečení fronty při další publikaci zprávy) se nejstarší zprávy začnou odstraňovat
x-overflow — tento argument se používá k nastavení chování v případě přetečení fronty. K dispozici jsou dvě hodnoty: drop-head (výchozí hodnota) a reject-publish. Pokud zvolíte drop-head, budou nejstarší zprávy odstraněny. Pokud zvolíte reject-publish, příjem zpráv bude pozastaven
x-dead-letter-exchange — určuje exchange, do kterého jsou směrovány odmítnuté zprávy, které nejsou znovu zařazeny do fronty
x-dead-letter-routing-key — určuje volitelný klíč pro směrování odmítnutých zpráv
x-max-priority — umožňuje třídění podle priorit ve frontě s maximální hodnotou priority 255 (RabbitMQ verze 3.5.0 a vyšší). Číslo udává maximální prioritu, kterou fronta podpoří. Pokud argument není nastaven, fronta nebude podporovat prioritu zpráv
x-queue-mode — umožňuje přepnout frontu do líného režimu. V tomto režimu bude co nejvíce zpráv uloženo na disku. Použití operační paměti bude minimální. Pokud není nastaven, fronta bude uchovávat zprávy v paměti, aby se zprávy doručovaly co nejrychleji
x-queue-master-locator — pokud máme cluster, lze nastavit master frontu
x-ha-policy — používá se při vytváření HA front a určuje, jak bude zpráva rozšířena mezi uzly. Pokud je nastavena hodnota all, zpráva bude uložena na všech uzlech. Pokud je nastavena hodnota nodes, zpráva bude uložena na konkrétních uzlech clusteru
x-ha-nodes — určuje uzly, k nimž bude patřit určitá fronta HA

Pokud je vytvoření fronty možné, pak server odešle klientovi synchronní RPC odpověď Queue.DeclareOk. Pokud je vytvoření fronty nemožné (došlo k odmítnutí žádosti Queue.Declare), pak kanál bude uzavřen serverem pomocí příkazu Channel.Close a klient obdrží výjimku OperationInterruptedException, která bude obsahovat kód chyby a její popis.

Opětovný příkaz Queue.Declare s podobnými parametry vrátí užitečné informace o této frontě. Například celkový počet zpráv čekajících ve frontě a celkový počet spotřebitelů přihlášených k ní.

Zavolání Queue.Declare pod uživatelskými údaji, kterému nebyla přiřazena potřebná oprávnění uzavře kanál pomocí příkazu Channel.Close a klient obdrží výjimku OperationInterruptedException, který bude obsahovat kód chyby 403 a jeho popis.

Jakmile fronta čeká déle než 10 sekund, přechází do spánkového režimu, vyvolávající GC ve frontě, což vede k výraznému snížení paměti potřebné pro tuto frontu.

Vytvoření fronty prostřednictvím grafického rozhraní

Přejdeme do administrátorské panelu RabbitMQ pod uživatelským jménem guest (uživatelské jméno: guest a heslo: guest). Všimněte si, že uživatel guest se může připojit pouze z místního hostitele. Nyní přejdeme na kartu Fronty a klikneme na Přidat novou frontu. Vyplňte vlastnosti:

Jakmile zadáme všechny potřebné údaje a stiskneme Přidat fronty, fronta se objeví v celkovém seznamu.

Kliknutí na název fronty zobrazí její podrobné informace. Zde lze nastavit vazbu mezi výměnou a frontou, zobrazit seznam spotřebitelé, publikovat/přijímat zprávy, odstranit frontu a zobrazit statistiky.

Vytvoření vazby

Vytvoření vazby probíhá pomocí synchronního RPC žádosti k serveru. Žádost je realizována pomocí metody Queue.Bind, volané s parametry:

název fronty
název bodu výměny
další parametry

Příklad vytvoření vazby pomocí RabbitMQ.Client:

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

fronta — název fronty
exchange — název výměníku
routingKey — klíč směrování
arguments — volitelné argumenty

Pokud bylo vytvoření vazby možné, pak server odešle klientovi synchronní RPC odpověď Queue.BindOk.

Vytvoření vazby prostřednictvím grafického rozhraní

Jakmile zadáme všechny potřebné údaje a stiskneme Bind, vazba se zobrazí v celkovém seznamu:

Kód

V této části popíšeme frontu a vazbu pomocí kódu v C#, jako bychom potřebovali vytvořit knihovnu. Může to být užitečné pro pochopení.

public interface IQueue
    {        
        string Name { get; }

        // 
        //     Pokud je nastavena true, fronta bude trvalá. 
        //     Bude uložena na disku a přežije 
        //     restart serveru/brokera. 
        //     Pokud je hodnota false, fronta je dočasná a bude smazána, 
        //     když bude server/broker restartován
        // 
        bool IsDurable { get; }

        // 
        //     Pokud je hodnota true, 
        //     tato fronta bude povolovat připojení 
        //     pouze jednomu consumeru
        // 
        bool IsExclusive { get; }

        // 
        //     Automatické odstranění. 
        //     Fronta bude odstraněna, když se všichni klienti odpojí.
        // 
        bool IsAutoDelete { get; }

        // 
        //     Volitelné argumenty
        // 
        IDictionary Arguments { get; }
    }

veřejná třída Queue : IQueue
    {
        veřejná třída Queue(
             řetězec jméno, 
             bool jeTrvalá = true, 
             bool jeExkluzivní = false, 
             bool jeAutomatickyMazána = false, 
             IDictionary<řetězec, objekt> argumenty = null)
        {
            Jméno = jméno ??
                vyhodit novou ArgumentNullException(jméno, $"{jméno} nesmí být null");

            JeTrvalá = jeTrvalá;
            JeExkluzivní = jeExkluzivní;
            JeAutomatickyMazána = jeAutomatickyMazána;
            Argumenty = argumenty ?? novou Dictionary<řetězec, objekt>();
        }

        veřejný řetězec Jméno { dostat; }
        veřejný bool JeTrvalá { dostat; }
        veřejný bool JeExkluzivní { dostat; }
        veřejný bool JeAutomatickyMazána { dostat; }
        veřejný IDictionary<řetězec, objekt> Argumenty { dostat; }
    }

veřejná statická třída QueueMode
    {       
        veřejná konstanta řetězec Výchozí = "výchozí";
        
        
        // <shrnout>
        //     Lenivý režim. Lenivý režim zajistí, že se uloží 
        //     co nejvíce zpráv na disk, aby se snížilo 
        //     využití operační paměti
        // </shrnout>
        veřejná konstanta řetězec Lenivý = "lenivý";
    }

veřejné rozhraní IBinding
    {
        // <shrnout>
        //     Výměnný systém, který bude spojovat vazbu
        // </shrnout>
        IExchange Exchange { dostat; }

        // <shrnout>
        //     Klíč pro směrování
        // </shrnout>
        řetězec RoutingKey { dostat; }

        // <shrnout>
        //     Nepovinné argumenty
        // </shrnout>
        IDictionary<řetězec, objekt> Argumenty { dostat; }
    }

veřejná třída Binding : IBinding
    {
        veřejná třída Binding(
             IExchange exchange, 
             řetězec routingKey, 
             IDictionary<řetězec, objekt> argumenty)
        {
            Exchange = exchange;
            RoutingKey = routingKey;
            Argumenty = argumenty;
        }

        veřejný IExchange Exchange { dostat; }
        veřejný řetězec RoutingKey { dostat; }
        veřejný IDictionary<řetězec, objekt> Argumenty { dostat; }
    }

Zdroj: habr.com