RabbitMQ. Partea 3. Înțelegem Queues și Bindings

Coadă (coada) — o structură de date pe disk sau în memorie care stochează referințe la mesaje și returnează copiile acestora consumers (consumatorilor). Coadă reprezintă proces Erlang cu starea (unde pot fi memorate și mesajele în sine). 1.000 de cozi pot ocupa aproximativ 80Mb.

Binding (legătură) — o regulă care informează brokerul în care dintre cozi ar trebui să ajungă mesajele.

Cuprins

• RabbitMQ. Partea 1. Introducere. Erlang, AMQP și RPC
• RabbitMQ. Partea 2. Să înțelegem schimburile
• RabbitMQ. Partea 3. Înțelegem Queues și Bindings
• RabbitMQ. Partea 4. Înțelegem ce reprezintă mesajele și cadrele
• RabbitMQ. Partea 5. Performanța publicării și consumului de mesaje
• RabbitMQ. Partea 6. Prezentarea modulelor Federation și Shovel
• RabbitMQ. Partea 7. Detalii despre Conexiune și Canal
• RabbitMQ. Partea 8. RabbitMQ în .NET
• RabbitMQ. Partea 9. Monitorizare

Cozi temporare

Dacă crearea cozii se face cu parametrul setat autoDelete, atunci această coadă devine capabilă să se ștergă automat. Astfel de cozi sunt de obicei create în momentul conectării primului client și sunt șterse atunci când toți clienții s-au deconectat.

Dacă crearea cozii se face cu parametrul setat exclusiv, atunci această coadă permit conectarea doar unui singur consumator și este ștearsă dacă canalul se închide. Până când canalul nu se închide, clientul poate să se deconecteze/conecteze, dar numai în cadrul aceleași conexiuni. Dacă parametrul exclusiv este setat, atunci parametrul autoDelete nu are niciun efect.

Caracteristici:

• în cazul unei întreruperi temporare a conexiunii, vom pierde mesajele care nu au ajuns încă la consumator
• se poate observa fenomenul binding churn. Fenomenul apare atunci când numărul de operațiuni de creare/ștergere a cozilor și legăturilor atinge valori foarte mari. În modul cluster, acest flux de operațiuni se va răspândi pe toate nodurile și va crea o încărcătură mare. Acest proces poate fi optimizat prin controlul numărului de subscrieri.

Cozi permanente

Dacă crearea cozii se face cu parametrul setat durabil, atunci această coadă își păstrează starea și se recuperează după repornirea serverului/brokerului. Această coadă va exista până când va fi invocată comanda Queue.Delete.

Cozi înalt disponibile

Cozi HA necesită un mediu cluster RabbitMQ. În modul cluster, toate informațiile despre exchange-uri, cozi, legături și consumatori vor fi copiate pe toate nodurile.

Când un mesaj este publicat într-o coadă HA, acesta este stocat pe fiecare nod care aparține cozii HA. După ce mesajul este consumat pe unul dintre noduri, toate copiile acestui mesaj vor fi șterse pe celelalte noduri.

Cozi HA pot fi distribuite pe toate nodurile dintr-un anumit cluster sau doar pe noduri individuale.

Caracteristici:

• utilizarea cozii HA duce la penalizări în performanță. Atunci când un mesaj este plasat într-o coadă HA sau este consumat din coada HA, RabbitMQ trebuie să coordoneze între toate nodurile (2-3 noduri sunt de obicei suficiente)

Crearea cozii

Crearea cozii se face printr-o operațiune sincronizată RPC cerere sincronă la server. Cererea se face prin metoda Queue.Declare, apelată cu următorii parametri:

• numele cozii
• alte parametrii

Exemplu de creare a cozii folosind RabbitMQ.Client:

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

• queue — numele cozii pe care dorim să o creăm. Numele trebuie să fie unic și nu poate coincide cu numele de sistem al cozii
• durabil — dacă este true, atunci coada va menține starea sa și se va restaura după repornirea serverului/broker-ului
• exclusiv — dacă este true, atunci coada va permite conectarea doar unui singur consumator
• autoDelete — dacă este true, atunci coada obține capacitatea să se ștergă automat
• arguments — argumente opționale. Mai jos vom detalia.

arguments

• x-message-ttl(x-message-time-to-live) — permite stabilirea timpului de expirare a mesajului în milisecunde. Dacă crearea cozii se face cu o valoare setată a argumentului x-message-ttl, atunci această coadă va excluzând automat mesajele a căror valoare de expirare a expirat. Setarea valorii argumentului x-message-ttl stabilește vârsta maximă pentru toate mesajele din această coadă. Crearea unei astfel de cozi permite prevenirea obținerii de informații învechite. Acesta poate fi utilizat în sistemele în timp real. Dacă coada pentru care este stabilit un schimb pentru mesajele respinse are setată valoarea argumentului x-message-ttl, atunci mesajele respinse din această coadă vor începe să aibă un termen de expirare.
• x-expires — stabilește o valoare în milisecunde după care coada este ștearsă. Coada poate să-și consume termenul de valabilitate doar dacă nu are abonați. Dacă sunt conectați abonați la coadă, aceasta se va putea șterge automat doar atunci când toți abonații vor apela Basic.Cancel sau se deconectează. Termenul de valabilitate al cozii se poate încheia doar dacă nu a fost cerut Basic.Get. Altfel, valoarea curentă a timpului de viață este resetată și coada nu va mai fi ștearsă automat. De asemenea, nu există garanții cu privire la cât de repede se va șterge coada după expirarea termenului său de valabilitate.
• x-max-length — stabilește numărul maxim de mesaje din coadă. Dacă numărul de mesaje din coadă începe să depășească limita maximă, cele mai vechi mesaje vor începe să fie șterse.

• x-max-lenght-bytes — stabilește dimensiunea maximă totală permisă a încărcăturii utile a mesajelor din coadă. Dacă valoarea stabilită este depășită (a apărut o supraplină a cozii la publicarea unui mesaj), cele mai vechi mesaje vor începe să fie șterse.
• x-overflow — acest argument este folosit pentru a configura comportamentul în urma unei supraplină a cozii. Sunt disponibile două valori: drop-head (valoare implicită) și reject-publish. Dacă alegeți drop-head, cele mai vechi mesaje vor fi șterse. Dacă alegeți reject-publish, recepția mesajelor va fi suspendată.
• x-dead-letter-exchange — stabilește exchange-ul în care sunt trimise mesajele respinse, care nu sunt reintroduse în coadă.
• x-dead-letter-routing-key — stabilește o cheie de rutare opțională pentru mesajele respinse.
• x-max-priority — permite sortarea pe baza priorităților în coadă, cu o valoare maximă a priorității de 255 (RabbitMQ versiunea 3.5.0 și ulterioare). Numărul indică prioritatea maximă pe care o va sprijini coada. Dacă argumentul nu este stabilit, coada nu va susține prioritățile mesajelor.
• x-queue-mode — permite trecerea coadei în mod leneș. În acest mod, cât mai multe mesaje vor fi stocate pe disc. Utilizarea memoriei RAM va fi minimă. În cazul în care nu este stabilit, coada va stoca mesajele în memorie pentru a livra mesajele cât mai rapid.
• x-queue-master-locator — dacă avem un cluster, atunci putem specifica coada master.
• x-ha-policy — este folosit la crearea coadelor HA și determină cum va fi distribuit un mesaj între noduri. Dacă este setată valoarea all, mesajul va fi păstrat pe toate nodurile. Dacă este setată valoarea nodes, mesajul va fi păstrat pe anumite noduri din cluster.
• x-ha-nodes — stabilește nodurile la care va fi legată o coadă. HA

Dacă crearea cozii se poate, serverul va trimite clientului o cerere sincronă RPC răspuns Queue.DeclareOk.Dacă crearea cozii nu este posibil (s-a produs o eroare în cerere Queue.Declare), atunci canalul se va închide serverului prin intermediul comenzii Channel.Close și clientul va primi o excepție OperationInterruptedException, care va conține codul de eroare și descrierea acesteia.

Un nou apel Queue.Declare cu parametrii similari va returna informații utile despre această coadă. De exemplu, numărul total de mesaje în așteptare în această coadă și numărul total de consumatori abonați la aceasta.

Apel Queue.Declare sub contul utilizatorului care nu are drepturile necesare va închide canalul folosind comanda Channel.Close și clientul va primi o excepție OperationInterruptedException, care va conține codul de eroare 403 și descrierea acesteia.

După ce coada stă în inactivitate timp de >= 10 secunde, aceasta intra în modul de așteptare, provocând apelarea GC în coadă, ceea ce duce la o reducere semnificativă a memoriei necesare pentru această coadă.

Crearea Queue prin interfața grafică

Accesăm panoul de administrare RabbitMQ sub utilizatorul guest (username: guest și password: guest). Rețineți că utilizatorul guest se poate conecta doar de pe gazda locală. Acum să mergem la tab-ul Queues și să facem clic pe Adăugați o nouă coadă. Completați proprietățile:

Odată ce am introdus toate datele necesare și apăsăm pe Adăugați cozi, coada va apărea în lista generală.

Click pe numele cozii va arăta informațiile detaliate. Aici putem configura legătura dintre schimb și coadă, vizualiza lista consumers, publica / primi mesaje, șterge coada și vizualiza statistici.

Crearea Binding

Crearea legăturii se face printr-un RPC cerere sincronă la server. Cererea se face prin metoda Queue.Bind, apelată cu următorii parametri:

• numele cozii
• numele punctului de schimb
• alte parametrii

Exemplu de creare a legăturii folosind RabbitMQ.Client:

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

• queue — numele cozii
• exchange — numele schimbului
• routingKey — cheia de rutare
• arguments — argumente opționale

Dacă crearea legăturii se poate, serverul va trimite clientului o cerere sincronă RPC răspuns Queue.BindOk.

Crearea Binding prin interfața grafică

Accesăm panoul de administrare RabbitMQ sub utilizatorul guest (username: guest și password: guest). Rețineți că utilizatorul guest se poate conecta doar de pe gazda locală. Acum să mergem la tab-ul Queues și apăsăm pe coadă my_queue. Completați câmpurile din secțiunea bindings:

Odată ce am introdus toate datele necesare și apăsăm pe Bind, legătura va fi afișată în lista generală:

Cod

În această secțiune vom descrie coada și legătura cu cod C#, ca și cum ar trebui să dezvoltăm o bibliotecă. Poate fi util pentru înțelegere.

public interface IQueue
    {        
        string Name { get; }

        // 

        //     Dacă este setat true, coada va fi persistentă. 
        //     Aceasta va fi stocată pe disc și va putea 
        //     supraviețui repornirii serverului / brokerului. 
        //     Dacă este false, coada este temporară și va fi ștearsă, 
        //     când serverul / brokerul va fi repornit
        // 
        bool IsDurable { get; }

        // 

        //     Dacă este true, 
        //     această coadă va permite conectarea 
        //     doar unui consumator
        // 
        bool IsExclusive { get; }

        // 

        //     Ștergere automată. 
        //     Coada va fi ștearsă, când toți clienții se deconectează.
        // 
        bool IsAutoDelete { get; }

        // 

        //     Argumente opționale
        // 
        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} must not be 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";
        // 

        //     Modul de lucru lenes. Modul lenes va forța salvarea 
        //     cât mai multor mesaje pe disc pentru a reduce 
        //     utilizarea memoriei operaționale
        // 
        public const string Lazy = "lazy";
    }

public interface IBinding
    {
        // 

        //     Schimbător care va fi legat printr-o legătură
        // 
        IExchange Exchange { get; }

        // 

        //     Cheie de rutare
        // 
        string RoutingKey { get; }

        // 

        //     Argumente opționale
        // 
        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; }
    }

Sursa: habr.com