🥇RabbitMQ. Del 3. Vi dykker ned i Queues og Bindings

Kø (kø) — en datastruktur på disk eller i arbejdshukommelsen, der opbevarer referencer til beskeder og udleverer deres kopier forbrugere (forbrugerne). Kø repræsenterer Erlang-proces med tilstand (hvor beskederne kan caches). 1.000 køer kan optage omkring 80Mb.

Binding (binding) — en regel, der informerer exchange om, hvilken kø beskederne skal sendes til.

Indholdsfortegnelse

RabbitMQ. Del 1. Introduktion. Erlang, AMQP og RPC
RabbitMQ. Del 2. Vi ser på Exchanges
RabbitMQ. Del 3. Vi ser på Queues og Bindings
RabbitMQ. Del 4. Vi ser på hvad beskeder og frames er
RabbitMQ. Del 5. Ydeevne for publicering og forbrug af beskeder
RabbitMQ. Del 6. Oversigt over modulerne Federation og Shovel
RabbitMQ. Del 7. Detaljer om Connection og Channel
RabbitMQ. Del 8. RabbitMQ i .NET
RabbitMQ. Del 9. Overvågning

Temporære køer

Hvis køen oprettes med den angivne parameter autoDelete, så får denne kø muligheden for automatisk at slette sig selv. Sådanne køer oprettes typisk, når den første klient opretter forbindelse og slettes, når alle klienter er frakoblet.

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

Egenskaber:

ved kortvarige forbindelsesafbrydelser vil vi miste beskeder, der endnu ikke er nået frem til forbrugeren
kan fænomenet binding churn. Fænomenet opstår, når antallet af operationer til oprettelse/fjernelse af køer og bindinger når meget store værdier. I klusterindstilling vil sådanne operationer sprede sig over alle noder og skabe en stor belastning. Denne proces kan optimeres ved at kontrollere antallet af tilmeldinger

Permanente køer

Hvis køen oprettes med den angivne parameter holdbar, så tillader denne kø bevarer sin tilstand og genoprettes efter genstart af serveren/broker. Denne kø vil eksistere, indtil kommandoen Queue.Delete.

Højt tilgængelige køer

HA-køer kræver en klynge RabbitMQ. I klyngeindstilling vil al information om exchanges, køer, bindinger og forbrugere blive kopieret til alle noder.

Når en besked publiceres i en HA-kø, gemmes den på hver node, der tilhører HA-køen. Når beskeden erstattes på en af noderne, vil alle kopier af denne besked blive slettet på de andre noder.

HA-køer kan strække sig over alle noder i et cluster eller kun enkelte.

Egenskaber:

Brugen af HA-køer medfører præstationsnedslag. Når en besked placeres i en HA-kø eller bliver forbrugt fra HA-køen, skal RabbitMQ koordinere mellem alle noder (2-3 noder er normalt tilstrækkeligt).

Oprettelse af kø

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

navnet på køen
andre parametre

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

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

kø – navnet på køen, vi ønsker at oprette. Navnet skal være unikt og må ikke være det samme som systemets kønavn.
holdbar – hvis true, vil køen bevare sin tilstand og genoprettes efter genstart af serveren/brokeren.
eksklusiv – hvis true, vil køen kun tillade, at én forbruger tilslutter sig.
autoDelete – hvis true, vil køen få evnen til for automatisk at slette sig selv
arguments – valgfrie argumenter. Vi vil gå nærmere ind på dem nedenfor.

arguments

x-message-ttl(x-message-time-to-live) – tillader at sætte udløbstid for beskeden i millisekunder. Hvis køen oprettes med en angivet argumentværdi, x-message-ttlvil en sådan kø automatisk udelukke beskeder, der er udløbet.At sætte værdien af argumentet x-message-ttl angiver den maksimale alder for alle beskeder i denne kø. Oprettelse af en sådan kø hjælper med at forhindre modtagelse af forældet information.Dette kan bruges i realtids-systemer. Hvis en kø, som har en exchange for afviste beskeder, får en angivet argumentværdi, x-message-ttlvil afviste beskeder i denne kø begynde at have en livstid.
x-expires — angiver værdien i millisekunder, hvorefter køen slettes. Køen kan kun udløbe, hvis den ikke har nogen abonnenter. Hvis der er abonnenter til køen, kan den kun automatisk slettes, når alle abonnenter har afsluttet forbindelsen. Basic.Cancel eller er frakoblet. Køens levetid kan kun udløbe, hvis der ikke har været nogen forespørgsel til den. Basic.Get. Ellers nulstilles den nuværende tidsindstillingsværdi, og køen vil ikke længere blive slettet automatisk. Der er heller ingen garantier for, hvor hurtigt køen slettes, efter dens levetid er udløbet. x-max-length.
— angiver det maksimale antal meddelelser i køen. Hvis antallet af meddelelser i køen overstiger det maksimale antal, vil de ældste meddelelser begynde at blive slettet. x-max-length-bytes

— angiver den maksimalt tilladte samlede størrelse af nyttelasten for meddelelserne i køen. Hvis den fastsatte værdi overskrides (der opstod en overfyldning af køen ved publicering af en meddelelse), vil de ældste meddelelser begynde at blive slettet. x-overflow
— dette argument bruges til at konfigurere adfærden i tilfælde af, at køen overfyldes. Der er to tilgængelige værdier: drop-head (standardværdi) og reject-publish . Hvis du vælger, vil de ældste meddelelser blive slettet. Hvis du vælger (standardværdi) og, vil modtagelse af meddelelser blive sat på pause. . Hvis du vælgerx-dead-letter-exchange
— angiver exchange, som de afviste meddelelser sendes til, som ikke sættes tilbage i køen. x-dead-letter-routing-key
— angiver en valgfri rute-nøgle for de afviste meddelelser. x-max-priority
— tillader sortering efter prioriteter i køen med en maksimal prioritet på 255 (RabbitMQ version 3.5.0 og nyere). Tallet angiver den maksimale prioritet, køen vil understøtte. Hvis argumentet ikke er indstillet, vil køen ikke understøtte meddelelsesprioriteter. x-queue-mode
— tillader at sætte køen i doven tilstand . I denne tilstand vil så mange meddelelser som muligt blive gemt på disken. Anvendelsen af RAM vil være minimal. Hvis det ikke er angivet, vil køen gemme meddelelser i hukommelsen for hurtigst mulige leverancer.x-queue-master-locator
— hvis vi har et cluster, kan en master kø angives. x-ha-policy
x-ha-policy — bruges ved oprettelse af HA-køer og definerer, hvordan beskeden vil blive distribueret til noderne. Hvis værdien er indstillet til alle, vil beskeden blive gemt på alle noder. Hvis værdien er indstillet til noder, vil beskeden blive gemt på bestemte noder i klyngen
x-ha-nodes — angiver de noder, som en given kø vil vedrøre HA

Hvis oprettelse af køen er muligt, vil serveren sende en synkron RPC @cassowary714 Queue.DeclareOk. Hvis oprettelse af køen ikke er muligt (anmodningen blev nægtet Queue.Declare), så vil kanalen blive lukket af serveren ved hjælp af kommandoen Channel.Close og klienten vil modtage en undtagelse OperationInterruptedException, som vil indeholde en fejlkode og en beskrivelse af fejlen.

Et gentaget 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 tilmeldte forbrugere.

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

Når køen har været inaktiv i >= 10 sekunder, går den i dvale, hvilket fremkalder GC i køen, hvilket resulterer i en betydelig reduktion af det nødvendige hukommelsesforbrug for denne kø.

Oprettelse af kø gennem den grafiske brugerflade

Gå til administrationspanelet RabbitMQ som bruger guest (brugernavn: guest og adgangskode: guest). Bemærk, at brugeren guest kun kan oprette forbindelse fra localhost. Nu går vi til fanen Queues og klikker på Add a new queue. Udfyld indstillingerne:

Når vi har indtastet alle nødvendige oplysninger og trykker på Add queues, vil køen blive tilføjet til den overordnede liste.

Klik på køens navn for at se dens detaljerede oplysninger. Her kan man konfigurere bindingen mellem udveksling og kø, se listen forbrugere, publicere/modtage beskeder, slette køen og se statistikken.

Oprettelse af binding

Binding skabes ved hjælp af en synkron RPC anmodning til serveren. Anmodningen udføres ved hjælp af metoden Queue.Bind, der kaldes med parametre:

navnet på køen
navnet på udvekslingspunktet
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å udveksleren
routingKey — nøglen til rutning
arguments — valgfri argumenter

Hvis oprettelsen af bindingen er muligt, vil serveren sende en synkron RPC @cassowary714 Queue.BindOk.

Oprettelse af binding gennem den grafiske brugerflade

Gå til administrationspanelet RabbitMQ som bruger guest (brugernavn: guest og adgangskode: guest). Bemærk, at brugeren guest kun kan oprette forbindelse fra localhost. Nu går vi til fanen Queues og tryk på køen my_queue. Udfyld felterne i sektionen bindings:

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

Kode

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

public interface IQueue
    {        
        string Name { get; }

        //
        //     Hvis sat til true, vil køen være permanent.
        //     Den vil blive gemt på disken og kan
        //     overleve genstart af server/broker.
        //     Hvis værdien er false, er køen midlertidig og vil blive slettet,
        //     når serveren/broker bliver genstartet
        //
        bool IsDurable { get; }

        //
        //     Hvis værdien er true,
        //     tillades kun én forbruger at tilslutte sig.
        //
        bool IsExclusive { get; }

        //
        //     Automatisk sletning.
        //     Køen vil blive slettet, når alle klienter 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";
        //
        //     Lazy mode. Lazy mode tvinger til at gemme
        //     så mange beskeder som muligt på disken for at reducere
        //     brugen af ram
        //
        public const string Lazy = "lazy";
    }

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

        //
        //     Routing nøgle
        //
        string RoutingKey { get; }

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

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

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

Kilde: habr.com