🥇RabbitMQ. Del 3. Vi ser på Queues og Bindings

Queue (kø) — en datastuktur på disken eller i RAM, der opbevarer referencer til meddelelser og leverer deres kopier forbrugere (til forbrugerne). Queue repræsenterer Erlang-proces med tilstand (hvor meddelelserne også kan cache-ses). 1.000 køer kan optage omkring 80 Mb.

Binding (binding) — en regel, der fortæller mægleren, hvilken kø meddelelserne skal sendes til.

Indholdsfortegnelse

RabbitMQ. Del 1. Introduktion. Erlang, AMQP og RPC
RabbitMQ. Del 2. Vi ser nærmere på Exchanges
RabbitMQ. Del 3. Vi ser nærmere på Queues og Bindings
RabbitMQ. Del 4. Vi ser nærmere på, hvad meddelelser og rammer er
RabbitMQ. Del 5. Ydelse for offentliggørelse og modtagelse af meddelelser
RabbitMQ. Del 6. Oversigt over modulerne Federation og Shovel
RabbitMQ. Del 7. Detaljeret om Connection og Channel
RabbitMQ. Del 8. RabbitMQ i .NET
RabbitMQ. Del 9. Overvågning

Tidligere køer

Hvis en kø oprettes med den indstillede parameter autoDelete, så får denne kø evnen til automatisk at slette sig selv. Sådanne køer oprettes typisk i det øjeblik, den første klient tilslutter, og slettes, når alle klienter disconnector.

Hvis en kø oprettes med den indstillede parameter eksklusiv, så tillader denne kø kun én forbruger at tilslutte og slettes, hvis kanalen lukkes. Indtil kanalen er lukket, kan klienten disconnecte/tilslutte, men kun inden for den samme forbindelse. Hvis parameteren eksklusiv er indstillet, så har parameteren autoDelete ingen effekt.

Egenskaber:

ved kortvarig forbindelse vil vi miste meddelelser, der endnu ikke er nået frem til forbrugeren
kan man opleve fænomenet binding churn. Fænomenet opstår, når antallet af operationer til oprettelse/sletning af køer og bindinger når meget høje værdier. I klynge-mode vil en sådan strøm af operationer sprede sig til alle noderne og skabe stor belastning. Denne proces kan optimeres gennem kontrol af antallet af abonneringer

Permanente køer

Hvis en kø oprettes med den indstillede parameter durable, så tillader denne kø bevarer sin tilstand og genopretter sig selv efter genstart af server/broker. Denne kø vil eksistere, så længe der ikke kaldes kommandoen Queue.Delete.

Højt tilgængelige køer

HA-køer kræver en klynge-miljø RabbitMQ. I klynge-mode vil al information om mæglerne, køerne, bindingerne og forbrugerne blive kopieret til alle noder.

Når en meddelelse offentliggøres i en HA-kø, gemmes den på hver node, der tilhører HA-køen. Når meddelelsen forbruges på en af noderne, vil alle kopier af denne meddelelse blive slettet på de andre noder.

HA-køer kan strække sig over alle noder i en given klynge eller kun på individuelle.

Egenskaber:

Brugen af HA-køer resulterer i en præstationsstraff. Når en meddelelse placeres i en HA-kø eller forbruges fra en HA-kø, skal RabbitMQ koordinere mellem alle noder (2-3 noder er normalt tilstrækkeligt).

Oprettelse af kø

Oprettelse af kø sker gennem en synkron RPC anmodning til serveren. Anmodningen foretages ved hjælp af metoden Queue.Declare, der kaldes med parametre:

køens navn
andre parametre

Eksempel på oprettelse af kø ved hjælp af RabbitMQ.Client:

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

kø — navnet på den kø, vi ønsker at oprette. Navnet skal være unikt og må ikke stemme overens med systemnavnet for køen.
durable — hvis true, vil køen bevare sin tilstand og gendannes efter genstart af serveren/mægleren.
eksklusiv — hvis true, vil køen kun tillade tilslutning fra én forbruger.
autoDelete — hvis true, vil køen blive i stand til automatisk at slette sig selv
arguments — valgfri argumenter. Lad os se nærmere på dem nedenfor.

arguments

x-message-ttl(x-message-time-to-live) — tillader at sætte udløbstidspunktet for meddelelsen i millisekunder. Hvis køen oprettes med den angivne argumentværdi x-message-ttl, vil en sådan kø automatisk udelukke meddelelser, der er udløbet.Indstillingen af argumentværdien x-message-ttl angiver den maksimale alder for alle meddelelser i denne kø. Oprettelsen af en sådan kø hjælper med at forhindre modtagelse af forældet information.Dette kan bruges i realtidsystemer. Hvis køen, for hvilken der er angivet en udveksling for afviste meddelelser, indstiller argumentværdien x-message-ttl, vil afviste meddelelser i denne kø begynde at have en levetid.
x-expires — angiver værdien i millisekunder, efter hvilken køen vil blive slettet. Køen kan kun nå sin udløbstid, hvis den ikke har nogen abonnenter. Hvis der er abonnenter til køen, kan den automatisk blive slettet, når alle abonnenter har kaldt Basic.Cancel eller frakoblet. Køens levetid kan kun udløbe, hvis der ikke har været nogen anmodninger til den Basic.Get. Ellers nulstilles den nuværende værdi af levetid, og køen vil ikke længere blive slettet automatisk. Desuden er der ingen garanti for, hvor hurtigt køen bliver slettet efter udløbet af dens levetid.
x-max-length — angiver det maksimale antal beskeder i køen. Hvis antallet af beskeder i køen begynder at overstige det maksimale antal, vil de ældste beskeder begynde at blive slettet

x-max-lenght-bytes — angiver den maksimalt tilladte samlede størrelse af nyttelasten af beskederne i køen. Når den angivne grænse overskrides (der opstod en overfyldning i køen ved næste offentliggørelse af en besked), vil de ældste beskeder begynde at blive slettet
x-overflow — denne parameter bruges til at kontrollere adfærden ved overfyldning af køen. To værdier er tilgængelige: drop-head (standardværdien) og reject-publish. Hvis du vælger drop-head, vil de ældste beskeder blive slettet. Hvis du vælger reject-publish, vil modtagelse af beskeder blive suspenderet
x-dead-letter-exchange — angiver den exchange, hvortil afviste beskeder sendes, som ikke genindsættes i køen
x-dead-letter-routing-key — angiver en valgfri rute nøgle for afviste beskeder
x-max-priority — tillader sortering efter prioriteter i køen med en maksimal prioriteringsværdi på 255 (RabbitMQ versioner 3.5.0 og senere). Tallet angiver den maksimale prioritet, som køen vil understøtte. Hvis argumentet ikke er indstillet, vil køen ikke understøtte beskedprioriteter
x-queue-mode — gør det muligt at føre køen i doven tilstand. I denne tilstand vil så mange beskeder som muligt blive gemt på disken. Brugen af RAM vil være minimal. Hvis den ikke er indstillet, vil køen gemme beskeder i hukommelsen for at levere dem så hurtigt som muligt
x-queue-master-locator — hvis vi har et klynge, kan vi angive master køen
x-ha-policy — anvendes ved oprettelse af HA-køer og definerer, hvordan beskeder distribueres på noderne. Hvis værdien er sat til alle, gemmes beskeden på alle noder. Hvis værdien er sat til noder, gemmes beskeden kun på bestemte noder i klyngen
x-ha-nodes — angiver de noder, som en bestemt kø vil høre til HA

Hvis oprettelse af køen er mulig, vil serveren sende en synkron RPC @cassowary714 Queue.DeclareOk. Hvis oprettelse af køen ikke er mulig (forespørgslen blev afvist Queue.Declare), så kanalen lukker af serveren med kommandoen Channel.Close og klienten vil modtage en undtagelse OperationInterruptedException, som vil indeholde fejlcode og en beskrivelse.

Et nyt kald Queue.Declare med de samme parametre vil returnere nyttige oplysninger om denne kø. For eksempel det samlede antal beskeder, der venter i denne kø, og det samlede antal forbrugere, der er tilmeldt.

Kald Queue.Declare med brugeroplysninger, der ikke har de nødvendige rettigheder vil lukke kanalen ved hjælp af kommandoen Channel.Close og klienten vil modtage en undtagelse OperationInterruptedException, der vil indeholde en fejlcode 403 og en beskrivelse.

Når køen er ledig i mere end >= 10 sekunder, går den i dvale, hvilket udløser GC i køen, hvilket resulterer i en betydelig reduktion af det nødvendige hukommelse til denne kø.

Oprettelse af kø via grafisk interface

Log ind på administratortavlen RabbitMQ under brugeren guest (brugernavn: guest og adgangskode: guest). Bemærk, at brugeren guest kun kan oprette forbindelse fra localhost. Lad os nu gå til fanen Queues og klikke på Add a new queue. Udfyld felterne:

Så snart vi indtaster alle nødvendige oplysninger og klikker på Add queues, vil køen dukke op i den generelle liste.

Et klik på køens navn viser dens detaljerede information. Her kan vi indstille bindingen mellem exchange og køen, se listen forbrugere, publicere/modtage beskeder, slette køen og se statistikker.

Oprettelse af Binding

Oprettelse af binding sker ved hjælp af synkron RPC anmodning til serveren. Anmodningen foretages ved hjælp af metoden Queue.Bind, der kaldes med parametre:

køens navn
navnet på exchange
andre parametre

Eksempel på oprettelse af binding ved hjælp af RabbitMQ.Client:

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

kø — navnet på køen
exchange — navnet på udvekslingen
routingKey — routing nøgle
arguments — valgfri argumenter

Hvis oprettelse af binding er mulig, vil serveren sende en synkron RPC @cassowary714 Queue.BindOk.

Oprettelse af Binding via grafisk interface

Når vi har indtastet alle nødvendige oplysninger og klikket på Bind, vil bindingen vises i den samlede liste:

Kode

I dette afsnit beskriver vi køen og bindingen med kode i C#, som hvis vi skulle udvikle et bibliotek. Dette kan være nyttigt for forståelsen.

public interface IQueue
    {        
        string Name { get; }

        // 
        //     Hvis true er indstillet, vil køen være vedholdende. 
        //     Den vil blive gemt på disken og kan 
        //     overleve genstart af server/broker. 
        //     Hvis false, er køen midlertidig og vil blive slettet, 
        //     når server/broker genstarter.
        // 
        bool IsDurable { get; }

        // 
        //     Hvis true, tillader denne kø kun én consumer at forbinde
        // 
        bool IsExclusive { get; }

        // 
        //     Automatisk sletning. 
        //     Køen vil blive slettet, når alle kunder er frakoblet.
        // 
        bool IsAutoDelete { get; }

        // 
        //     Valgfri argumenter
        // 
        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} må ikke være 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";
        // 
        //     Længe mode. Længe mode tvinger til at gemme 
        //     så mange meddelelser som muligt på disken for at reducere 
        //     brugen af RAM
        // 
        public const string Lazy = "lazy";
    }

public interface IBinding
    {
        // 
        //     Udveksling, der vil blive bundet med binding
        // 
        IExchange Exchange { get; }

        // 
        //     Rutingnøgle
        // 
        string RoutingKey { get; }

        // 
        //     Valgfri argumenter
        // 
        IDictionary Arguments { get; }
    }

offentlig klasse Binding : IBinding
    {
        offentlig Binding(
             IExchange exchange, 
             streng routingKey, 
             IDictionary argumenter)
        {
            Exchange = exchange;
            RoutingKey = routingKey;
            Arguments = argumenter;
        }

        offentlig IExchange Exchange { get; }
        offentlig streng RoutingKey { get; }
        offentlig IDictionary Arguments { get; }
    }

Kilde: habr.com