🥇RabbitMQ. Osa 3. Uurime Queues ja Bindings

Oote (oote) — andmestruktuur, mis asub kettal või mälus ja hoiab linke sõnumitele ning edastab nende koopiaid tarbijatele (tarbijatele). Oote on Erlang-protsess olekuga (kus võivad olla vahemälus ka sõnumid). 1000 ootejärjekorda võib võtta umbes 80Mb.

Sidumine (sidumine) — reegel, mis ütleb vahetusele, millisesse ootejärjekorda sõnumid peaksid jõudma.

Sisukord

RabbitMQ. Osa 1. Sissejuhatus. Erlang, AMQP ja RPC
RabbitMQ. Osa 2. Uurime välja Exchanges
RabbitMQ. Osa 3. Uurime Queue-sid ja Bindings-e
RabbitMQ. Osa 4. Uurime, mis on sõnumid ja raami
RabbitMQ. Osa 5. Sõnumite publikatsiooni ja tarbimise tõhusus
RabbitMQ. Osa 6. Ülevaade modulaarsete Federation ja Shovel
RabbitMQ. Osa 7. Üksikasjalik teave Connection ja Chaneli kohta
RabbitMQ. Osa 8. RabbitMQ .NET-is
RabbitMQ. Osa 9. Jälgimine

Ajutised ootejärjekorrad

Kui ootejärjekorra loomine toimub määratud parameetri autoDelete, siis selline järjekord omandab võime automaatne kustutamine. Sellised järjekorrad luuakse tavaliselt esimese kliendi ühendamisel ja kustutatakse siis, kui kõik kliendid lahti ühenduvad.

Kui ootejärjekorra loomine toimub määratud parameetri eksklusiivne, siis selline järjekord lubab ainult ühel tarbijal ühenduda ja kustutatakse, kui kanal suletakse. Kuni kanal ei ole suletud, võib klient lahti ühenduda ja uuesti ühenduda, kuid ainult sama seose piires. Kui parameeter eksklusiivne on määratud, siis parameeter autoDelete ei oma mingit mõju.

suurendatud ressursside piirangud;

ajutise katkestuse korral kaotame sõnumid, mis ei ole veel tarbijani jõudnud
võime jälgida nähtust sidumise keerukus. See nähtus tekib, kui loodud ja eemaldatud ootejärjekordade ja sidumiste arv saavutab väga suure väärtuse. Klasterrežiimis levib see toimingute voog kõigi sõlmede vahel ja loob suurt koormust. Seda protsessi saab optimeerida tellimuste arvu kontrollimise kaudu

Püsivad ootejärjekorrad

Kui ootejärjekorra loomine toimub määratud parameetri kestvad, siis selline järjekord salvestavad oma oleku ja taastuvad pärast serveri/brokkeri taaskäivitamist. Antud järjekord püsib seni, kuni käivitatakse käsk Queue.Delete.

Kõrge kättesaadavusega ootejärjekorrad

HA järjekorrad nõuavad RabbitMQ klastrikeskkonda. Klusterrežiimis kopeeritakse kogu teave vahetuste, ootejärjekordade, sidumiste ja tarbijate kohta kõigile sõlmedele.

Kui sõnum avaldatakse HA järjekorda, salvestatakse see iga sõlme, mis kuulub HA järjekonda. Pärast seda, kui sõnum tarbitakse mõnes sõlmes, eemaldatakse kõik selle sõnumi koopiad teistelt sõlmedelt.

HA järjekorrad võivad laieneda kõigile sõlmedele mingis klastris või ainult individuaalsetele.

suurendatud ressursside piirangud;

HA järjekordade kasutamine võib kaasa tuua tulemuslikkuse karistusi. Kui sõnum pannakse mingisse HA järjekorda või kui sõnumit tarbitakse HA järjekorrast, peab RabbitMQ koordineerima kõikide sõlmede vahel (2-3 sõlme on tavaliselt piisavad).

Järjekorra loomine

Järjekorra loomine toimub sünkroonse RPC serveri päringu kaudu. Päring tehakse meetodi Queue.Declare, mis kutsutakse esile koos parameetritega:

järjekorra nimi
muu parameeter

Näide järjekorra loomisest kasutades RabbitMQ.Client:

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

queue — järjekorra nimi, mille soovime luua. Nimi peab olema ainulaadne ja ei tohi kattuda süsteemi järjekorra nimega.
kestvad — kui true, siis järjekord säilitab oma oleku ja taastatakse pärast serveri/brokkeri taaskäivitamist. — kui true, siis järjekord lubab ühendada ainult ühte tarbijat.
eksklusiivne — kui true, siis järjekord omandab võime
autoDelete arguments automaatne kustutamine
— valikulised argumendid. Allpool vaatame lähemalt. x-message-ttl

— valikulised argumendid. Allpool vaatame lähemalt.

x-message-time-to-live() — võimaldab määrata sõnumi eluea millisekundites. Kui järjekorra loomine toimub seatud argumendi väärtusega,, siis selline järjekord x-message-time-to-liveautomaatselt kõrvaldada sõnumid, mille kehtivusaeg on lõppenud. Argumendi väärtuse seadminemäärab maksimaalse vanuse kõigile nende järjekorra sõnumitele. Sellise järjekorra loomine x-message-time-to-live aitab vältida aegunud teabe saamist. Seda saab kasutada reaalajas süsteemides. Kui järjekord, mille jaoks on seatud vahetiga äraviskatud sõnumid, seadistatakse argumendi väärtus, siis äraviskatud sõnumid selles järjekorras x-message-time-to-livealgavad eluea saamist. x-expires.
x-präägib — määrab millisekundide arvu, mille möödudes toimub järjekorra kustutamine. Järjekord võib oma eluiga ära kasutada ainult siis, kui tal ei ole mingeid tellijaid. Kui järjekorrale on ühendatud tellijad, saab see automaatselt kustuda ainult siis, kui kõik tellijad kutsuvad Basic.Cancel või lahti ühendavad. Järjekorra eluea lõppemine võib toimuda ainult juhul, kui selle vastu ei ole esitatud nõudmisi Basic.Get. Vastasel juhul lähtestatakse praegune eluaja seadistus ja järjekord ei kustuta enam automaatselt. Samuti ei ole garantiid, kui kiiresti järjekord kustutatakse pärast selle eluaja lõppemist.
x-max-length — määrab maksimaalse sõnumite arvu järjekorras. Kui sõnumite arv järjekorras hakkab ületama maksimaalset arvu, hakkavad eemalduma vanimad

x-max-length-bytes — määrab maksimaalselt lubatud koormuse suuruse järjekorras. Üksuse ületamisel (kui järjekord ületati järgmise sõnumi avaldamise ajal) hakkavad vanimad sõnumid eemalduma
x-overflow — see argument kasutatakse järjekorra ületamise käitumise seadmiseks. Saadaval on kaks väärtust: drop-head (vaikimisi väärtus) ja reject-publish. Kui valida drop-head, siis eemaldatakse vanimad sõnumid. Kui valida reject-publish, siis sõnumite vastuvõtt peatatakse
x-dead-letter-exchange — määrab vahetuse, kuhu suunatakse tagasilükatud sõnumid, mis ei ole uuesti järjekorda pandud
x-dead-letter-routing-key — määrab valikulise marsruudivõtme tagasilükatud sõnumite jaoks
x-max-priority — võimaldab järjekorras prioriteetide järgi sortimist maksimaalse prioriteedi väärtusega 255 (RabbitMQ versioonides 3.5.0 ja kõrgem). Number näitab maksimaalset prioriteeti, mida järjekord toetab. Kui argument puudub, ei toeta järjekord sõnumite prioriteete
x-queue-mode — võimaldab viia järjekorra laiskrežiimi. Sellises režiimis hoitakse võimalikult palju sõnumeid kettal. Mälu kasutamine on minimaalne. Kui see ei ole seadistatud, salvestab järjekord sõnumid mällu, et edastada sõnumeid maksimaalse kiirus
x-queue-master-locator — kui meil on klaster, saab määrata pea järjekorra
x-ha-policy — kasutatakse HA järjekordade loomisel ja määrab, kuidas sõnum levib sõlmedes. Kui väärtus on seadistatud all, siis sõnum säilitakse kõigis sõlmedes. Kui väärtus on seadistatud sõlmed, siis sõnum säilitakse teatud klastrisõlmedes
x-ha-nodes — määrab sõlmed, millele teatud järjekord kuulub HA

Kui järjekorra loomine võib-olla, siis server saadab kliendile sünkroonse RPC vastuse Queue.DeclareOk. Kui järjekorra loomine teostada (tekib tõrge päringus Queue.Declare), siis kanal suletakse serveri poolt käsuga Channel.Close ja klient saab erandi OperationInterruptedException, mis sisaldab veakoodi ja selle kirjelduse.

Uuesti kutsumine Queue.Declare sarnaste parameetritega tagastab kasulikku informatsiooni selle järjekorra kohta. Näiteks, kokku sõnumite arv, mis ootavad selle järjekorra peal, ja kokku registreeritud tarbijaid.

Kutsumine Queue.Declare kasutaja, kellele ei ole antud vajalikke õigusi sulgeb kanali käskluse abil Channel.Close ja klient saab erandi OperationInterruptedException, mis sisaldab veakoodi 403 ja selle kirjelduse.

Pärast seda, kui järjekord on olnud vaikne > 10 sekundit, kimbu seisu, kutsudes GC järjekorras, mis toob kaasa märkimisväärse mälumahtude vähenemise, mis on vajalik selle järjekorra jaoks.

Järjekorra loomine graafilise liidese kaudu

Siseneme administraatori paneeli RabbitMQ kasutaja all guest (username: guest ja password: guest). Pange tähele, et kasutaja guest saab ühenduda ainult lokaalsest hostist. Nüüd liigutame vahekaardile Queues ja vajutame Lisa uus järjekord. Täidame omadused:

Kui oleme sisestanud kõik vajalikud andmed ja vajutame Lisa järjekorrad, siis järjekord ilmub üldnimekirja.

Nimele klõpsamine näitab selle detailselt teavet. Siin saab seada sidumist vahetuse ja järjekorra vahel, vaadata nimekirja tarbijatele, avaldada/saada sõnumeid, kustutada järjekord ja vaadata statistikat.

Sidumise loomine

Sidumise loomine toimub sünkroonse kaudu RPC serveri päringu kaudu. Päring tehakse meetodi Queue.Bind, mis kutsutakse esile koos parameetritega:

järjekorra nimi
vahetuspunkti nimi
muu parameeter

Näide sidumise loomisest kasutatakse RabbitMQ.Client:

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

queue — järjekorra nimi
vahetus — vahetuspunkti nimi
routingKey — marsruudi võti
— valikulised argumendid. Allpool vaatame lähemalt. — valikulised argumendid

Kui sidumise loomine võib-olla, siis server saadab kliendile sünkroonse RPC vastuse Queue.BindOk.

Sidumise loomine graafilise liidese kaudu

Kui oleme sisestanud kõik vajalikud andmed ja vajutame Siduma, seondus kuvatakse üldises loetelus:

Kood

Käesolevas jaotises kirjeldame järjekorda ja sidumise koodi C#-s, justkui peaksime arendama teeki. See võib osutuda kasulikuks arusaamiseks.

public interface IQueue
    {        
        string Name { get; }

        // 
        //     Kui on seatud true, siis queue on püsiv. 
        //     See salvestatakse kettale ja suudab 
        //     ellu jääda serveri/brokeri taaskäivitamisel. 
        //     Kui väärtus on false, siis queue on ajutine ja kustutatakse, 
        //     kui server/broker taaskäivitub
        // 
        bool IsDurable { get; }

        // 
        //     Kui väärtus on true, siis 
        //     see järjekord lubab ühenduda 
        //     ainult ühel tarbijal
        // 
        bool IsExclusive { get; }

        // 
        //     Automaatne kustutamine. 
        //     Järjekord kustutatakse, kui kõik kliendid lahti ühenduvad.
        // 
        bool IsAutoDelete { get; }

        // 
        //     Valikulised argumendid
        // 
        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} ei tohi olla 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";
        // 
        //     Laisk režiim. Laisk režiim sunnib säilitama 
        //     nii palju sõnumeid kettal kui võimalik, et vähendada 
        //     mälu kasutamist
        // 
        public const string Lazy = "lazy";
    }

public interface IBinding
    {
        // 
        //     Vahetus, millega sidumine toimub
        // 
        IExchange Exchange { get; }

        // 
        //     Marsruutimisvõti
        // 
        string RoutingKey { get; }

        // 
        //     Valikulised argumendid
        // 
        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; }
    }

Allikas: habr.com