🥇RabbitMQ. Osa 3. Tutvume järjekordade ja sidumistega

Queue (järjekord) — andmestruktuur, mis asub kettal või mälus ning salvestab viiteid sõnumitele ning edastab nende koopiaid consumers (tarbijatele). Queue esindab Erlang-protsessi olekus (kuhu võivad vahemällu salvestuda ka sõnumid). 1000 järjekorda võib võtta umbes 80Mb.

Binding (sidumine) — reegel, mis ütleb vahetusele, millisesse järjekorda sõnumid peavad minema.

Sisukord

RabbitMQ. Osa 1. Sissejuhatus. Erlang, AMQP ja RPC
RabbitMQ. Osa 2. Uurime Exchanges
RabbitMQ. Osa 3. Uurime Queues ja Bindings
RabbitMQ. Osa 4. Uurime, mis on sõnumid ja raamid
RabbitMQ. Osa 5. Sõnumite avaldamise ja tarbimise jõudlus
RabbitMQ. Osa 6. Ülevaade moodulitest Federation ja Shovel
RabbitMQ. Osa 7. Üksikasjalikult Connection ja Channel
RabbitMQ. Osa 8. RabbitMQ .NET-is
RabbitMQ. Osa 9. Juhendamine

Ajutised järjekorrad

Kui järjekorra loomine toimub seadistatud parameetriga autoDelete, siis selline järjekord saab võime automaatne end ära kustutada. Sellised järjekorrad loodi tavaliselt siis, kui esimene klient on ühendatud, ja kustutatakse hetkedel, kui kõik kliendid on lahti ühendatud.

Kui järjekorra loomine toimub seadistatud parameetriga exclusive, siis selline järjekord lubab ainult ühele tarbijale ühenduma ja eemaldatakse, kui kanal suletakse. Kuni kanali sulgemiseni võib klient lahti ühenduda / taas ühendada, kuid ainult sama ühenduse piires. Kui parameeter exclusive on seadistatud, siis parameeter autoDelete ei avalda mingit mõju.

Omadused:

lühiajaliste ühenduste katkestuste korral kaotame sõnumid, mis ei ole veel tarbijani jõudnud
võib kogeda fenomeni binding churn. See fenomen tekib siis, kui järjekordade ja sidumiste loomise / kustutamise operatsioonide hulk saavutab väga suuri väärtusi. Klasterrežiimis laienevad sellised operatsioonivoogud kõikidele sõlmedele ja genereerivad suurt koormust. Selle protsessi saab optimeerida tellimuste arvu kontrollimisega

Püsivad järjekorrad

Kui järjekorra loomine toimub seadistatud parameetriga durable, siis selline järjekord säilitavad oma oleku ja taastuvad pärast serveri / vahendaja taaskäivitamist. See järjekord eksisteerib kuni korratakse käsku Queue.Delete.

Kõrge saadavuse järjekorrad

HA järjekorrad nõuavad RabbitMQ klusterkeskkonda. Klasterrežiimis kopeeritakse kogu teave vahetite, järjekordade, sidumiste ja tarbijate kohta kõigile sõlmedele.

Kui sõnum avaldatakse HA järjekorda, salvestatakse see igas sõlmes, mis kuulub HA järjekorda. Kui sõnum tarbitakse mõnes sõlmes, siis kõik selle sõnumi koopiad kustutatakse teistest sõlmedest.

HA järjekorrad võivad laieneda kõikidesse sõlmedesse mingis klastris või ainult üksikutesse.

Omadused:

HA järjekordade kasutamine toob kaasa jõudluse karistusi. Kui sõnum paigutatakse mõnesse HA järjekorda või kui sõnum tarbitakse HA järjekorrast, peab RabbitMQ koordineerima kõiki sõlmi (2-3 sõlme on tavaliselt piisavad).

Järjekorra loomine

Järjekorra loomine toimub sünkrone RPC päringuga serverisse. Päring tehakse meetodi kaudu Queue.Declare, mida kutsutakse välja parameetritega:

järjekorra nimi
teised parameetrid

Näide järjekorra loomisest RabbitMQ.Client:

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

queue — järjekorra nimi, mida soovime luua. Nimi peab olema ainulaadne ja ei tohi kattuda süsteemse järjekorra nimega.
durable — kui true, siis järjekord salvestab oma seisundi ja taastub pärast serveri/brokri taaskäivitamist.
exclusive — kui true, siis järjekord lubab ühendada ainult ühte tarbijat
autoDelete — kui true, siis järjekord omandab võime automaatne end ära kustutada
argumentide — valikulised argumendid. Allpool käsitleme seda lähemalt.

argumentide

x-message-ttl(x-message-time-to-live) — võimaldab seadistada sõnumi eluea kestust millisekundites. Kui järjekorra loomine toimub seatud argumentidega x-message-ttl, siis selline järjekord kätkeb automaatselt sõnumeid, mille kehtivusaeg on lõppenud. Argumendi seadmise väärtus x-message-ttl määrab maksimaalse vanuse kõikidele sõnumitele antud järjekorras. Sellise järjekorra loomine aitab vältida aegunud teabe saamist. Seda saab kasutada reaalajas süsteemides. Kui järjekord, millele on määratud vahetus tagasilükatud sõnumite jaoks, seada argumendi väärtus x-message-ttl, siis tagasilükatud sõnumid antud järjekorras hakkavad omama eluea.
x-expires — määrab väärtuse millisekundites, mille möödumisel järjekord kustutatakse. Järjekord võib oma kehtivuse tähtaega kasutada ainult siis, kui sellel ei ole tellijaid. Kui järjekorrale on ühendatud tellijad, saab see automaatselt kustutada, ainult kui kõik tellijad väljakutse esitavad Basic.Cancel või lahti ühendavad. Järjekorra eluaeg saab lõppeda ainult juhul, kui sellele ei ole esitatud päringut Basic.Get. Vastasel juhul lähtestatakse praegune eluaegade seade nulliks ja järjekord ei kustutata enam automaatselt. Samuti ei ole garanteeritud, kui kiiresti järjekord kustutatakse pärast selle eluea lõppemist.
x-max-length — määrab maksimaalse sõnumite arvu järjekorras. Kui sõnumite arv järjekorras hakkab ületama maksimaalset arvu, siis kustutatakse kõige vanemad

x-max-lenght-bytes — määrab maksimaalse lubatud koormuse sõnumite summaarsuuruse järjekorras. Kui seatud väärtust ületatakse (järjekorra ületäide tekkis järgmise sõnumi avaldamise ajal), hakkavad kõige vanemad sõnumid kustutama
x-overflow — seda argumenti kasutatakse järjekorra ülevoolu käitumise seadistamiseks. Saadaval on kaks väärtust: drop-head (vaikevää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 exchange'i, kuhu suunatakse tagasi lükatud sõnumid, mida ei panna uuesti järjekorda
x-dead-letter-routing-key — määrab valikulise suunamisvõtme tagasi lükatud sõnumitele
x-max-priority — lubab sorteerimise prioriteetide alusel järjekorras maksimaalse prioriteedi väärtusega 255 (RabbitMQ versioonides 3.5.0 ja uuemates). Number näitab maksimaalset prioriteeti, mida järjekord toetab. Kui argument pole seadistatud, ei toeta järjekord sõnumite prioriteeti
x-queue-mode — võimaldab viia järjekorra laisk režiim. Sellises režiimis hoitakse võimalikult palju sõnumeid kettal. Operatiivmälu kasutamine on minimaalne. Kui seda ei ole seadistatud, hoiab järjekord sõnumeid mälu, et edastada sõnumeid võimalikult kiiresti
x-queue-master-locator — kui meil on klaster, siis saab seada meistrijärjekorra
x-ha-policy — kasutatakse HA järjestuste loomiseks ja määrab, kuidas sõnum jaguneb sõlmede vahel. Kui väärtuseks on seatud all, siis säilitatakse sõnum kõigil sõlmedel. Kui väärtuseks on seatud nodes, siis säilitatakse sõnum määratud klastrisõlmedel
x-ha-nodes — määrab sõlmed, millele teatav järjekord kuulub HA

Kui järjekorra loomine on võimalik, siis server saadab kliendile sünkroonse RPC vastuse Queue.DeclareOk. Kui järjekorra loomine ei ole võimalik (taotluse tõttu tekkis tõrge Queue.Declare), siis kanal suletakse serveri poolt käsuga Channel.Close ja klient saab erandi OperationInterruptedException, mis sisaldab veakoodi ja selle kirjeldust.

Uuesti kutsumine Queue.Declare samade parameetritega tagastab kasulikku teavet selle järjekorra kohta. Näiteks sõnumite koguarv, mis ootavad antud järjekorras, ja registreeritud tarbijate koguarv.

Kutse Queue.Declare kasutaja mandaatide all, kellele ei ole määratud vajalikke õigusi sulgeb kanali käsuga Channel.Close ja klient saab erandi OperationInterruptedException, mis sisaldab veakoodi 403 ja selle kirjeldust.

Pärast seda, kui järjekord on olnud 10 sekundi jooksul inaktiivne, sattub see ootusrežiimi, kutsudes esile GC järjekorras, mis viib olulise mälu kokkuhoidmiseni, mis on vajalik sellele järjekorrale.

Queue'i loomine graafilise liidese kaudu

Siseneme administraatori paneeli RabbitMQ kasutaja all guest (kasutajanimi: guest ja parool: guest). Pange tähele, et kasutaja guest saab ühendust luua ainult kohalikust hostist. Nüüd liigume vahekaardile Queues ja klõpsame Add a new queue. Täidame omadused:

Koheselt pärast vajutamist Add queues, ilmub järjekord üldnimekirja.

Järjekorra nimele klõpsamine näitab selle üksikasjalikku teavet. Siit saab seadistada sideme vahetuse ja järjekorra vahel, vaadata loendit consumers, avaldada/saada sõnumeid, kustutada järjekorda ja vaadata statistikat.

Sideme loomine

Sideme loomine toimub sünkroneeritud RPC päringuga serverisse. Päring tehakse meetodi kaudu Queue.Bind, mida kutsutakse välja parameetritega:

järjekorra nimi
vahetuse punkti nimi
teised parameetrid

Näidis sideme loomise kohta RabbitMQ.Client:

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

queue — järjekorra nimi
exchange — vahetuse nimi
routingKey — marsruudivõti
argumentide — valikuline argument

Kui sideme loomine on võimalik, siis server saadab kliendile sünkroonse RPC vastuse Queue.BindOk.

Sideme loomine graafilise liidese kaudu

Niipea, kui anname kõik vajalikud andmed ja vajutame Siduda, kuvatakse seos üldises loendis:

Kood

Selles jaotises kirjeldame järjekorda ja sidumist C# koodiga, nagu oleksime arendanud raamatukogu. See võib olla kasulik arusaamiseks.

public interface IQueue
    {        
        string Name { get; }

        // <summary>
        //     Kui seadistada true, siis järjekord on püsiv. 
        //     See salvestatakse kettale ja suudab 
        //     taluda serveri/brokeri taasalustamist. 
        //     Kui väärtus on false, siis järjekord on ajutine ja kustutatakse, 
        //     kui server/broker taasalustatakse
        // </summary>
        bool IsDurable { get; }

        // <summary>
        //     Kui väärtus on true, siis 
        //     sellesse järjekorda tohib ühendada 
        //     vaid üks consumer
        // </summary>
        bool IsExclusive { get; }

        // <summary>
        //     Automaatne kustutamine. 
        //     Järjekord kustutatakse, kui kõik kliendid lahkuvad.
        // </summary>
        bool IsAutoDelete { get; }

        // <summary>
        //     Valikuline argument
        // </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} ei tohi olla tühi");

            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 salvestama 
        //     võimalikult palju sõnumeid kettale, et vähendada 
        //     mälu kasutamist
        // 
        public const string Lazy = "lazy";
    }

public interface IBinding
    {
        // 
        //     Vahetus, mis seondub sidumisega
        // 
        IExchange Exchange { get; }

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

        // 
        //     Valikuline argument
        // 
        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