🥇RabbitMQ. 3. del. Razumevanje čakalnih vrst in vezav

Queue (čakalna vrsta) - podatkovna struktura na disku ali v RAM-u, ki shranjuje sklice na sporočila in pošilja njihove kopije consumers (potrošnikom). Queue je a Erlang proces s stanjem (kjer se lahko predpomnijo sporočila). 1 tisoč čakalnih vrst lahko zasede približno 80 Mb.

Binding (binding) - pravilo, ki izmenjevalniku pove, v katero čakalno vrsto naj gredo sporočila.

Kazalo

RabbitMQ. 1. del. Uvod. Erlang, AMQP in RPC
RabbitMQ. 2. del. Razumevanje izmenjav
RabbitMQ. Del 3. Razumevanje čakalnih vrst in vezav
RabbitMQ. 4. del. Razumevanje sporočil in okvirjev
RabbitMQ. 5. del: Zmogljivost objavljanja in uživanja sporočil
RabbitMQ. Del 6. Pregled modulov Federation in Shovel
RabbitMQ. Del 7. Podrobnosti o Connection in Chanel
RabbitMQ. Del 8. RabbitMQ v .NET
RabbitMQ. Del 9. Spremljanje

Začasne čakalne vrste

Če pride do ustvarjanja čakalne vrste z naborom parametrov autoDelete, potem taka čakalna vrsta pridobi sposobnost se samodejno izbrišete. Takšne čakalne vrste se običajno ustvarijo, ko se prvi odjemalec poveže, in se izbrišejo, ko vsi odjemalci prekinejo povezavo.

Če pride do ustvarjanja čakalne vrste z naborom parametrov exclusive, potem taka čakalna vrsta omogoča priklop le enega porabnika in se izbriše, če je kanal zaprt. Dokler se kanal ne zapre, se odjemalec lahko odklopi/priklopi, vendar le znotraj iste povezave. Če je parameter exclusive je nastavljen, nato parameter autoDelete nima učinka.

Značilnosti:

v primeru kratkotrajne prekinitve povezave bomo izgubili sporočila, ki še niso dosegla potrošnika
lahko ujamete pojav binding churn. Do pojava pride, ko število operacij za ustvarjanje/brisanje čakalnih vrst in povezav doseže zelo velike vrednosti. V načinu gruče se bo tak tok operacij razširil na vsa vozlišča in ustvaril veliko obremenitev. Ta postopek je mogoče optimizirati z nadzorom števila naročnin

Nenehne čakalne vrste

Če pride do ustvarjanja čakalne vrste z naborom parametrov durable, potem taka čakalna vrsta ohranja svoje stanje in se obnovi po ponovnem zagonu strežnika/posrednika. Ta čakalna vrsta bo obstajala do klica ukaza Queue.Delete.

Zelo razpoložljiva čakalna vrsta

Čakalne vrste HA zahtevajo okolje gruče RabbitMQ. V načinu gruče bodo vse informacije o izmenjevalcih, čakalnih vrstah, vezavah in porabnikih kopirane v vsa vozlišča.

Ko je sporočilo objavljeno v čakalni vrsti HA, je shranjeno v vsakem vozlišču, ki pripada čakalni vrsti HA. Ko je sporočilo zaužito v enem vozlišču, bodo vse kopije tega sporočila izbrisane v drugih vozliščih.

Čakalne vrste HA so lahko razdeljene na vsa vozlišča v določeni gruči ali samo na posamezna.

Značilnosti:

uporaba čakalnih vrst HA ima za posledico kazni za zmogljivost. Pri postavitvi sporočila v čakalno vrsto HA ali porabi sporočila iz čakalne vrste HA se mora RabbitMQ uskladiti med vsemi vozlišči (2-3 vozlišča so običajno dovolj)

Ustvarjanje čakalne vrste

Čakalna vrsta je ustvarjena s sinhronim RPC zahtevo strežniku. Zahteva se izvede z metodo Queue.Declareklicano s parametri:

ime čakalne vrste
drugi parametri

Primer ustvarjanja čakalne vrste z uporabo RabbitMQ.Client:

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

queue — ime čakalne vrste, ki jo želimo ustvariti. Ime mora biti unikatno in ne sme biti enako sistemskemu imenu čakalne vrste
durable — če je res, potem bo čakalna vrsta rešiti svoje bogastvo in se obnovi po ponovnem zagonu strežnika/posrednika
exclusive — če je res, bo čakalna vrsta dovolila samo enemu potrošniku povezavo
autoDelete — če je res, potem čakalna vrsta pridobi sposobnost se samodejno izbrišete
arguments - neobvezni argumenti. Oglejmo si ga podrobneje spodaj.

Argumenti

x-message-ttl(x-message-time-to-live) - omogoča nastavitev časa poteka sporočila v milisekundah. Če pride do ustvarjanja čakalne vrste z nastavljeno vrednostjo argumenta x-message-ttl, potem bo takšna čakalna vrsta samodejno izloči sporočila, ki so potekla. Nastavitev vrednosti argumenta x-message-ttl določa največjo starost za vsa sporočila v tej čakalni vrsti. Ustvarjanje takšne čakalne vrste pomaga preprečiti prejemanje zastarelih informacij. To se lahko uporablja v sistemih v realnem času. Če je čakalna vrsta, za katero je nastavljen izmenjevalec za zavrnjena sporočila, nastavite vrednost argumenta x-message-ttl, je nato zavrnil sporočila v tej čakalni vrsti bo začelo imeti življenjsko dobo.
x-expires — nastavi vrednost v milisekundah, po kateri se čakalna vrsta izbriše. Čakalna vrsta lahko poteče le, če nima naročnikov. Če so v čakalno vrsto priključeni naročniki, se le-ta lahko samodejno izbriše šele, ko pokličejo vsi naročniki Basic.Cancel ali prekinite povezavo. Čakalna vrsta lahko poteče samo, če zanjo ni bila vložena nobena zahteva Basic.Get. V nasprotnem primeru se vrednost nastavitve trenutne življenjske dobe ponastavi na nič in čakalna vrsta ne bo več samodejno izbrisana. tudi ni nobenih zagotovil, kako hitro bo čakalna vrsta izbrisana po izteku njene življenjske dobe.
x-max-length — nastavi največje število sporočil v čakalni vrsti. Če število sporočil v čakalni vrsti začne presegati največje število, se najstarejša začnejo brisati

x-max-lenght-bytes — nastavi največjo dovoljeno skupno velikost sporočil v čakalni vrsti. Če je nastavljena vrednost presežena (čakalna vrsta se prepolni ob naslednji objavi sporočila), se začnejo brisati najstarejša sporočila.
x-overflow — ta argument se uporablja za konfiguracijo vedenja kot rezultat prepolnitve čakalne vrste. Na voljo sta dve vrednosti: drop-head (privzeta vrednost) in reject-publish. Če izberete drop-head, bodo najstarejša sporočila izbrisana. Če izberete reject-publish, potem bo prejemanje sporočil prekinjeno
x-dead-letter-exchange — določa izmenjavo, v katero se pošljejo zavrnjena sporočila, ki niso ponovno v čakalni vrsti
x-dead-letter-routing-key — določa izbirni usmerjevalni ključ za zavrnjena sporočila
x-max-priority — omogoča razvrščanje po prioriteti v čakalni vrsti z največjo prednostno vrednostjo 255 (RabbitMQ različice 3.5.0 in višje). Številka označuje največjo prioriteto, ki jo podpira čakalna vrsta. Če argument ni nastavljen, čakalna vrsta ne bo podpirala prioritete sporočil
x-queue-mode — omogoča prenos čakalne vrste na leni način. V tem načinu bo na disku shranjenih čim več sporočil. Poraba RAM-a bo minimalna. Če ni nastavljena, bo čakalna vrsta shranjevala sporočila v pomnilnik za čim hitrejšo dostavo sporočil
x-queue-master-locator — če imamo gručo, potem lahko nastavimo glavno čakalno vrsto
x-ha-policy — uporablja se pri ustvarjanju čakalnih vrst HA in določa, kako bo sporočilo razdeljeno med vozlišča. Če je nastavljeno na all, potem bo sporočilo shranjeno na vseh vozliščih. Če je nastavljeno na nodes, potem bo sporočilo shranjeno v določenih vozliščih gruče
x-ha-nodes — določa vozlišča, na katera se bo določena čakalna vrsta nanašala HA

Če ustvarite čakalno vrsto morda, potem bo strežnik odjemalcu poslal sinhrono RPC odgovor Queue.DeclareOk. Če ustvarite čakalno vrsto nemogoče (zahteva je bila zavrnjena Queue.Declare) torej kanal se bo zaprl strežnik z uporabo ukaza Channel.Close in stranka bo prejela izjemo OperationInterruptedException, ki bo vseboval kodo napake in njen opis.

Odpoklic Queue.Declare s podobnimi parametri bo vrnil koristne informacije o tej čakalni vrsti. Na primer skupno število čakajočih sporočil v določeni čakalni vrsti in skupno število uporabnikov, ki so nanjo naročeni.

Izziv Queue.Declare pod poverilnicami uporabnika, ki mu niso bile dodeljene potrebne pravice bo zaprl kanal z uporabo ukaza Channel.Close in stranka bo prejela izjemo OperationInterruptedException, ki bo vseboval kodo napake 403 in njegov opis.

Ko je čakalna vrsta nedejavna >= 10 sekund, se preide v način mirovanjas klicem GC v čakalni vrsti, kar ima za posledico znatno zmanjšanje pomnilnika, potrebnega za to čakalno vrsto.

Ustvarjanje čakalne vrste prek GUI

Pojdite na skrbniško ploščo RabbitMQ pod uporabnikom guest (uporabniško ime: guest in geslo: guest). Upoštevajte, da uporabnik guest se lahko poveže le z lokalnega gostitelja. Zdaj pa pojdimo na zavihek Queues in kliknite na Add a new queue. Izpolnite lastnosti:

Takoj, ko vnesemo vse potrebne podatke in kliknemo na Add queues, bo čakalna vrsta prikazana na splošnem seznamu.

S klikom na ime čakalne vrste se prikažejo podrobne informacije o njej. Tukaj lahko konfigurirate vezavo med centralo in čakalno vrsto, glejte seznam consumers, objavljanje/prejemanje sporočil, brisanje čakalne vrste in ogled statistike.

Ustvarjanje vezave

Vezava je ustvarjena s sinhronim RPC zahtevo strežniku. Zahteva se izvede z metodo Queue.Bindklicano s parametri:

ime čakalne vrste
ime menjalne točke
drugi parametri

Primer ustvarjanja vezave z uporabo RabbitMQ.Client:

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

queue - ime čakalne vrste
exchange — naziv menjalnika
routingKey - usmerjevalni ključ
arguments - neobvezni argumenti

Če ustvarjate vezavo morda, potem bo strežnik odjemalcu poslal sinhrono RPC odgovor Queue.BindOk.

Ustvarjanje vezave prek GUI

Ko vnesemo vse zahtevane podatke in kliknemo na Bind, bo vezava prikazana na splošnem seznamu:

Koda

V tem razdelku bomo opisali čakalno vrsto in vezavo s kodo C#, kot da bi morali razviti knjižnico. Morda bo to koristno za zaznavanje.

public interface IQueue
    {        
        string Name { get; }

        /// <summary>
        ///     Если установить true, то queue будет являться постоянным. 
        ///     Она будет храниться на диске и сможет 
        ///     пережить перезапуск сервера/брокера. 
        ///     Если значение false, то queue является временной и будет удаляться, 
        ///     когда сервер/брокер будет перезагружен
        /// </summary>
        bool IsDurable { get; }

        /// <summary>
        ///     Если значение равно true, то 
        ///     такая очередь будет разрешать подключаться 
        ///     только одному consumer-у
        /// </summary>
        bool IsExclusive { get; }

        /// <summary>
        ///     Автоматическое удаление. 
        ///     Очередь будет удалена, когда все клиенты отсоединятся.
        /// </summary>
        bool IsAutoDelete { get; }

        /// <summary>
        ///     Необязательные аргументы
        /// </summary>
        IDictionary<string, object> Arguments { get; }
    }

public class Queue : IQueue
    {
        public Queue(
             string name, 
             bool isDurable = true, 
             bool isExclusive = false, 
             bool isAutoDelete = false, 
             IDictionary<string, object> arguments = null)
        {
            Name = name ??
                throw new ArgumentNullException(name, $"{name} must not be null");

            IsDurable = isDurable;
            IsExclusive = isExclusive;
            IsAutoDelete = isAutoDelete;
            Arguments = arguments ?? new Dictionary<string, object>();
        }

        public string Name { get; }
        public bool IsDurable { get; }
        public bool IsExclusive { get; }
        public bool IsAutoDelete { get; }
        public IDictionary<string, object> Arguments { get; }
    }

public static class QueueMode
    {       
        public const string Default = "default";
        /// <summary>
        ///     Ленивый режим. Ленивый режим заставит сохранять 
        ///     как можно больше сообщений на диске, чтобы уменьшить 
        ///     использование оперативной памяти
        /// </summary>
        public const string Lazy = "lazy";
    }

public interface IBinding
    {
        /// <summary>
        ///     Обменник, который будет связываться привязкой
        /// </summary>
        IExchange Exchange { get; }

        /// <summary>
        ///     Ключ маршрутизации
        /// </summary>
        string RoutingKey { get; }

        /// <summary>
        ///     Необязательные аргументы
        /// </summary>
        IDictionary<string, object> Arguments { get; }
    }

public class Binding : IBinding
    {
        public Binding(
             IExchange exchange, 
             string routingKey, 
             IDictionary<string, object> arguments)
        {
            Exchange = exchange;
            RoutingKey = routingKey;
            Arguments = arguments;
        }

        public IExchange Exchange { get; }
        public string RoutingKey { get; }
        public IDictionary<string, object> Arguments { get; }
    }

Vir: www.habr.com