🥇RabbitMQ. Deel 3. We bestuderen Queues en Bindings

Wachtrij (wachtrij) - een datastructuur op schijf of in het geheugen die verwijzingen naar berichten opslaat en deze kopieën levert consumers (consumenten). Wachtrij is Erlang-proces met een status (waar berichten zelf kunnen worden cached). 1000 wachtrijen kunnen ongeveer 80Mb in beslag nemen.

Binding (binding) - een regel die de broker vertelt naar welke wachtrijen berichten moeten worden gestuurd.

Inhoudsopgave

RabbitMQ. Deel 1. Introductie. Erlang, AMQP en RPC
RabbitMQ. Deel 2. Laten we Exchanges verkennen
RabbitMQ. Deel 3. Laten we kijken naar Queues en Bindings
RabbitMQ. Deel 4. Laten we kijken wat berichten en frames zijn
RabbitMQ. Deel 5. Prestatie van het publiceren en consumeren van berichten
RabbitMQ. Deel 6. Overzicht van de modules Federation en Shovel
RabbitMQ. Deel 7. Gedetailleerd over Connection en Channel
RabbitMQ. Deel 8. RabbitMQ in .NET
RabbitMQ. Deel 9. Monitoring

Tijdelijke wachtrijen

Als het aanmaken van een wachtrij gebeurt met een ingestelde parameter autoDelete, dan krijgt die wachtrij het vermogen zichzelf automatisch te verwijderen. Dergelijke wachtrijen worden meestal aangemaakt op het moment dat de eerste client verbinding maakt en worden verwijderd op het moment dat alle clients zijn losgekoppeld.

Als het aanmaken van een wachtrij gebeurt met een ingestelde parameter exclusief, dan staat die wachtrij slechts één consument toe om verbinding te maken en wordt verwijderd als de kanaal sluit. Tot dat moment kan de client loskoppelen/aansluiten, maar alleen binnen dezelfde verbinding. Als de parameter exclusief is ingesteld, dan heeft de parameter autoDelete geen effect.

Kenmerken:

bij een tijdelijke onderbreking van de verbinding zullen we berichten verliezen die nog niet bij de consument zijn aangekomen
kan men het fenomeen binding churnvastleggen. Het fenomeen ontstaat wanneer het aantal aanmaak/verwijderoperaties van wachtrijen en bindingen zeer hoge waarden bereikt. In cluster-modus zal deze stroom van operaties zich verspreiden over alle knooppunten en een hoge belasting creëren. Dit proces kan worden geoptimaliseerd door het aantal abonnementen te controleren.

Permanente wachtrijen

Als het aanmaken van een wachtrij gebeurt met een ingestelde parameter duurable, dan staat die wachtrij behouden hun status en herstellen zich na een herstart van de server/broker. Deze wachtrij zal bestaan totdat het commando Queue.Delete.

Highly Available wachtrijen

HA-wachtrijen vereisen een clustered RabbitMQ-omgeving. In cluster-modus zal alle informatie over brokers, wachtrijen, bindingen en consumenten worden gekopieerd naar alle knooppunten.

Wanneer een bericht in een HA-wachtrij wordt gepubliceerd, wordt het op elk knooppunt opgeslagen dat bij de HA-wachtrij hoort. Nadat het bericht op een van de knooppunten is geconsumeerd, worden alle kopieën van dat bericht op andere knooppunten verwijderd.

HA-wachtrijen kunnen zich over alle knooppunten in een cluster verspreiden of alleen over specifieke.

Kenmerken:

Het gebruik van HA-queues leidt tot prestatienadelen. Bij het plaatsen van een bericht in een HA-queue of het consumeren van een bericht uit een HA-queue, moet RabbitMQ coördinatie uitvoeren tussen alle knooppunten (2-3 knooppunten zijn meestal voldoende)

Een queue creëren

Het creëren van een queue gebeurt synchronisch RPC verzoek aan de server. Het verzoek wordt gedaan met behulp van de methode Queue.Declare, aangeroepen met de parameters:

de naam van de queue
andere parameters

Voorbeeld van het creëren van een queue met behulp van RabbitMQ.Client:

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

queue — de naam van de queue die we willen creëren. De naam moet uniek zijn en mag niet overeenkomen met een systeemnaam van een queue
duurable — als true, zal de queue zijn status behouden en hersteld worden na een herstart van de server/broker
exclusief — als true, zal de queue toestaan dat alleen één consument verbinding maakt
autoDelete — als true, krijgt de queue de mogelijkheid zichzelf automatisch te verwijderen
arguments — optionele argumenten. Hieronder bespreken we dit verder.

arguments

x-message-ttl(x-message-time-to-live) — stelt de levensduur van een bericht in milliseconden in. Als de queue wordt gecreëerd met een ingesteld argumentwaarde x-message-ttl, dan zal die queue automatisch berichten uitsluiten waarvan de termijn is verstreken. Het instellen van de argumentwaarde x-message-ttl stelt de maximale leeftijd in voor alle berichten in deze queue. Het creëren van zo'n queue helpt om verouderde informatie te voorkomen. Dit kan worden gebruikt in realtime-systemen. Als er voor een queue waaraan een wisselaar voor afgewezen berichten is toegewezen, een argumentwaarde wordt ingesteld x-message-ttl, dan zullen de afgewezen berichten in deze queue een levensduur hebben.
x-expires — stelt de waarde in milliseconden in waarbij de queue wordt verwijderd. Een queue kan zijn levensduur alleen verbruiken als deze geen abonnees heeft. Als er abonnees aan de queue zijn verbonden, kan deze automatisch worden verwijderd zodra alle abonnees Basic.Cancel of losgekoppeld zijn. De levensduur van de queue kan alleen eindigen als er geen verzoek naar is geweest Basic.Get. Anders wordt de huidige levensduur instelling gereset en zal de queue niet langer automatisch worden verwijderd. Ook is er geen garantie over hoe snel de verwijdering van de queue plaatsvindt na het verstrijken van de levensduur.
x-max-length — stelt het maximum aantal berichten in de wachtrij in. Als het aantal berichten in de wachtrij het maximum overschrijdt, zullen de oudste berichten worden verwijderd.

x-max-lenght-bytes — stelt de maximaal toegestane totale grootte van de nuttige lading van berichten in de wachtrij in. Bij overschrijding van de ingestelde waarde (ontstond een overflow in de wachtrij bij een volgende publicatie van een bericht) zullen de oudste berichten worden verwijderd.
x-overflow — dit argument wordt gebruikt om het gedrag bij een overflow van de wachtrij in te stellen. Er zijn twee waarden beschikbaar: drop-head (standaard waarde) en reject-publish. Als u kiest voor drop-head, worden de oudste berichten verwijderd. Als u kiest voor reject-publish, wordt de acceptatie van berichten tijdelijk stopgezet.
x-dead-letter-exchange — stelt de exchange in waar afgewezen berichten naar worden gestuurd, die niet opnieuw in de wachtrij worden geplaatst.
x-dead-letter-routing-key — stelt een optionele routeringssleutel in voor afgewezen berichten.
x-max-priority — staat sortering op prioriteit toe in de wachtrij met een maximale prioriteit van 255 (RabbitMQ versies 3.5.0 en hoger). Het getal geeft de maximale prioriteit aan die de wachtrij zal ondersteunen. Als het argument niet is ingesteld, ondersteunt de wachtrij geen prioriteitsberichten.
x-queue-mode — maakt het mogelijk om de wachtrij in te stellen op luie modus. In deze modus zullen zoveel mogelijk berichten op de schijf worden opgeslagen. Het gebruik van het RAM zal minimaal zijn. Als het niet is ingesteld, zal de wachtrij berichten in het geheugen opslaan om berichten zo snel mogelijk te leveren.
x-queue-master-locator — als we een cluster hebben, kan de master wachtrij worden ingesteld.
x-ha-policy — wordt gebruikt bij het maken van HA wachtrijen en bepaalt hoe een bericht over de knooppunten wordt verspreid. Als de waarde is ingesteld op all, wordt het bericht op alle knooppunten opgeslagen. Als de waarde is ingesteld op nodes, wordt het bericht op bepaalde knooppunten van het cluster opgeslagen.
x-ha-nodes — stelt de knooppunten in waartoe een bepaalde wachtrij zal behoren. HA

Als de wachtrij is gemaakt mogelijk, zal de server een synchronisatie RPC op de kritiek van het gebruik van advertentieblokker als middel om persoonlijke conflicten op te lossen en druk uit te oefenen op gebruikers, verklaarde ontwikkelaar dimisa Queue.DeclareOk. Als de wachtrij is gemaakt onmogelijk is (er was een fout bij het verzoek Queue.Declare), dan zal de kanaal sluiten door de server met het command Channel.Close en de client ontvangt een uitzondering OperationInterruptedException, die de foutcode en de omschrijving zal bevatten.

Een herroep Queue.Declare met dezelfde parameters zal nuttige informatie over deze wachtrij teruggeven. Bijvoorbeeld, het totale aantal berichten dat wacht in deze wachtrij en het totale aantal consumenten dat zich op deze wachtrij heeft geabonneerd.

Aanroep Queue.Declare onder gebruikersgegevens waarvoor niet de nodige rechten zijn toegewezen sluit het kanaal met behulp van het commando Channel.Close en de client ontvangt een uitzondering OperationInterruptedException, dat de foutcode bevat 403 en de beschrijving ervan.

Nadat de wachtrij gedurende >= 10 seconden heeft gerust, valt deze in de slaapstand, waardoor GC in de wachtrij wordt aangeroepen, wat leidt tot een aanzienlijke vermindering van het geheugen dat nodig is voor deze wachtrij.

Een wachtrij aanmaken via de grafische interface

Ga naar het beheerderspaneel RabbitMQ onder gebruikersnaam gast; (username: gast; en password: gast;). Let op dat de gebruiker gast; alleen vanaf de lokale host kan verbinden. Laten we nu naar het tabblad Wachtrijen gaan en op de Voeg een nieuwe wachtrij toe. Vul de eigenschappen in:

Zodra we alle nodige gegevens hebben ingevoerd en op Wachtrijen toevoegen, verschijnt de wachtrij in de algemene lijst.

Een klik op de naam van de wachtrij toont de gedetailleerde informatie. Hier kan de binding tussen de uitwisseling en de wachtrij worden ingesteld, kan de lijst worden bekeken consumers, berichten worden gepubliceerd/ontvangen, de wachtrij kan worden verwijderd en de statistieken kunnen worden bekeken.

Binding aanmaken

Het aanmaken van een binding gebeurt via een synchrone RPC verzoek aan de server. Het verzoek wordt gedaan met behulp van de methode Queue.Bind, aangeroepen met de parameters:

de naam van de queue
naam van het uitwisselpunt
andere parameters

Voorbeeld van het aanmaken van een binding met behulp van RabbitMQ.Client:

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

queue — naam van de wachtrij
exchange — naam van de uitwisseling
routingKey — routeringssleutel
arguments — optionele argumenten

Als de binding wordt aangemaakt mogelijk, zal de server een synchronisatie RPC op de kritiek van het gebruik van advertentieblokker als middel om persoonlijke conflicten op te lossen en druk uit te oefenen op gebruikers, verklaarde ontwikkelaar dimisa Queue.BindOk.

Binding aanmaken via de grafische interface

Ga naar het beheerderspaneel RabbitMQ onder gebruikersnaam gast; (username: gast; en password: gast;). Let op dat de gebruiker gast; alleen vanaf de lokale host kan verbinden. Laten we nu naar het tabblad Wachtrijen en klikken op de wachtrij my_queue. Vul de velden in het gedeelte in bindings:

Zodra we alle nodige gegevens hebben ingevoerd en op Bind, wordt de binding weergegeven in de algemene lijst:

Code

In dit gedeelte beschrijven we de wachtrij en binding met C#-code, alsof we een bibliotheek zouden moeten ontwikkelen. Dit kan nuttig zijn voor het begrip.

public interface IQueue
    {        
        string Name { get; }

        // <summary>
        //     Als true is ingesteld, is de wachtrij permanent. 
        //     Deze wordt op schijf opgeslagen en kan 
        //     herstarten van de server/broker overleven. 
        //     Als false, is de wachtrij tijdelijk en wordt verwijderd, 
        //     wanneer de server/broker opnieuw wordt opgestart
        // </summary>
        bool IsDurable { get; }

        // <summary>
        //     Als true is ingesteld, 
        //     mag deze wachtrij alleen door één consumer worden benaderd
        // </summary>
        bool IsExclusive { get; }

        // <summary>
        //     Automatische verwijdering. 
        //     De wachtrij wordt verwijderd wanneer alle clients worden ontkoppeld.
        // </summary>
        bool IsAutoDelete { get; }

        // <summary>
        //     Optionele argumenten
        // </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} mag niet null zijn");

            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";
        // 
        //     Luie modus. De luie modus zorgt ervoor dat er zoveel mogelijk berichten op de schijf worden opgeslagen om het gebruik van het geheugen te minimaliseren
        // 
        public const string Lazy = "lazy";
    }

public interface IBinding
    {
        // 
        //     Wisselaar die een binding zal uitvoeren
        // 
        IExchange Exchange { get; }

        // 
        //     Routeringssleutel
        // 
        string RoutingKey { get; }

        // 
        //     Optionele argumenten
        // 
        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; }
    }

Bron: habr.com