🥇RabbitMQ. Del 3. Utforska Queues och Bindings

Kö (kö) — en datstruktur på disken eller i RAM som lagrar referenser till meddelanden och returnerar deras kopior consumers (till konsumenterna). Kö representerar Erlang-process med tillstånd (där meddelandena själva kan cachelagras). 1 tusen köer kan ta upp omkring 80 Mb.

Binding (bindning) — en regel som informerar meddelandebrokern om vilken kö meddelanden ska gå till.

Innehållsförteckning

RabbitMQ. Del 1. Introduktion. Erlang, AMQP och RPC
RabbitMQ. Del 2. Vi går igenom Exchanges
RabbitMQ. Del 3. Vi går igenom Queues och Bindings
RabbitMQ. Del 4. Låt oss förstå vad meddelanden och ramar är
RabbitMQ. Del 5. Prestanda för publicering och konsumtion av meddelanden
RabbitMQ. Del 6. Översikt av modulerna Federation och Shovel
RabbitMQ. Del 7. Detaljer om Connection och Channel
RabbitMQ. Del 8. RabbitMQ i .NET
RabbitMQ. Del 9. Övervakning

Tillfälliga köer

Om köcreationen sker med en inställd parameter autoDelete, så får den köen möjlighet att automatiskt ta bort sig själv. Dessa köer skapas vanligtvis vid anslutningen av den första klienten och tas bort när alla klienter har kopplats bort.

Om köcreationen sker med en inställd parameter exklusiv, så tillåter den kö att endast en konsument kan ansluta och tas bort om kanalen stängs. Tills kanalen stängs kan klienten koppla bort/ansluta, men bara inom samma anslutning. Om parametern exklusiv är inställd, så har parametern autoDelete ingen effekt.

Egenskaper:

vid kortvarigt avbrott i kommunikationen kommer vi att förlora meddelanden som ännu inte har nått konsumenten
kan fånga fenomenet bindingskrusning. Fenomenet uppstår när antalet operationer för att skapa/ta bort köer och bindningar når mycket höga nivåer. I klusterläge kommer denna ström av operationer att spridas över alla noder och skapa en stor belastning. Denna process kan optimeras genom att kontrollera antalet prenumerationer

Permanenta köer

Om köcreationen sker med en inställd parameter hållbar, så tillåter den kö bevarar sitt tillstånd och återställs efter omstart av servern/brokern. Denna kö kommer att existera tills kommandot Queue.Delete.

Högt tillgängliga köer

HA-köer kräver en klustermiljö för RabbitMQ. I klusterläge kommer all information om växlare, köer, bindningar och konsumenter att kopieras till alla noder.

När ett meddelande publiceras i en HA-kö, lagras det på varje nod som tillhör HA-kön. När meddelandet konsumeras på någon av noderna kommer alla kopior av meddelandet att tas bort från andra noder.

HA-köer kan spridas över alla noder i ett kluster eller bara på individuella.

Egenskaper:

Användning av HA-köer leder till straff i prestanda. När ett meddelande placeras i en HA-kö eller när ett meddelande konsumeras från en HA-kö, måste RabbitMQ samordna mellan alla noder (2-3 noder är vanligtvis tillräckliga).

Skapa kö

Köning skapande sker med hjälp av synkron. RPC begäran till servern. Begäran görs genom metoden Queue.Declare, som anropas med följande parametrar:

Kö namn
andra parametrar

Exempel på kö skapande med hjälp av RabbitMQ.Client:

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

queue — namnet på kö som vi vill skapa. Namnet måste vara unikt och får inte koincidera med systemnamnet för kö.
hållbar — om true, kommer kön att spara sitt tillstånd och återställas efter en omstart av servern/mäklaren.
exklusiv — om true, kommer kön att tillåta endast en konsument att ansluta.
autoDelete — om true, ger kön möjlighet att automatiskt ta bort sig själv
argument — valfria argument. Nedan går vi in mer detaljerat.

argument

x-message-ttl(x-message-time-to-live) — gör det möjligt att ställa in meddelandets livslängd i millisekunder. Om kö skapas med ett angivet värde för argumentet x-message-ttl, så kommer en sådan kö att automatiskt exkludera meddelanden vars livslängd har gått ut.Att ställa in värdet för argumentet x-message-ttl anger maxåldern för alla meddelanden i denna kö. Att skapa en sådan kö kan förhindra att föråldrad information tas emot.Detta kan användas i realtidssystem. Om en kö har ett utbyte för avvisade meddelanden och får värdet för argumentet x-message-ttl, kommer avvisade meddelanden i denna kö att börja ha en livslängd..
x-expires — anger ett värde i millisekunder efter vilket kön tas bort. En kö kan utlöpa sitt liv endast om den inte har några prenumeranter. Om det finns prenumeranter ansluten till kön, kan den automatiskt tas bort endast när alla prenumeranter anropar Basic.Cancel eller kopplar bort sig. En kö livslängd kan endast avslutas om den inte har begärts. Basic.Get. Annars nollställs nuvarande livslängdsinställning, och kön kommer inte längre att tas bort automatiskt. Även finns det inga garantier för hur snabbt kön tas bort efter att dess livslängd har löpt ut..
x-max-length — anger maxantal meddelanden i kön. Om antalet meddelanden i kön börjar överskrida det maximala antalet, kommer de äldsta börja rensas bort.

x-max-längd-bytes — anger den maximalt tillåtna sammanlagda storleken av nyttolasten i meddelandena i kön. Om det angivna värdet överskrids (en överbelastning i kön inträffar vid en ny publicering av ett meddelande) kommer de äldsta meddelandena att börja rensas bort.
x-overflow — detta argument används för att ställa in beteendet vid överflöd i kön. Två värden är tillgängliga: drop-head (standardvärdet) och reject-publish. Om du väljer drop-head, kommer de äldsta meddelandena att rensas bort. Om du väljer reject-publish, kommer mottagandet av meddelanden att stoppas.
x-dead-letter-exchange — anger en exchange dit avvisade meddelanden skickas, som inte har ställts tillbaka i kön.
x-dead-letter-routing-key — anger en valfri routningsnyckel för avvisade meddelanden.
x-max-priority — möjliggör sortering efter prioritet i kön med en maximal prioritet på 255 (RabbitMQ versioner 3.5.0 och senare). Siffran anger den maximala prioriteten som kön kommer att stödja. Om argumentet inte anges kommer kön inte att stödja meddelandets prioritet.
x-queue-mode — tillåter att köen överförs till lågprioritetsläget. I detta läge kommer så många meddelanden som möjligt att sparas på disken. Användningen av RAM-minne kommer att vara minimal. Om det inte anges kommer kön att spara meddelanden i minnet för att leverera meddelanden så snabbt som möjligt.
x-queue-master-locator — om vi har en kluster kan vi ange masterkön.
x-ha-policy — används vid skapandet av HA-köer och definierar hur meddelanden kommer att spridas över noder. Om värdet anges all, kommer meddelandet att sparas på alla noder. Om värdet anges nodes, kommer meddelandet att sparas på specifika noder i klustret.
x-ha-nodes — anger noder som en viss kö kommer att tillhöra. HA

Om skapandet av kön är möjligt, så kommer servern att skicka ett synkront RPC svar Queue.DeclareOk. Om skapandet av kön är omöjligt (det har skett ett avslag på begäran Queue.Declare), så kanalen stängs av servern med kommandot Channel.Close och klienten får ett undantag OperationInterruptedException, som kommer att innehålla felkoden och dess beskrivning.

Ett återanrop Queue.Declare med samma parametrar kommer att returnera användbar information om denna kö. Till exempel det totala antalet meddelanden som väntar i denna kö och det totala antalet prenumeranter på den.

Anrop Queue.Declare med användaruppgifter som inte har de nödvändiga rättigheterna tilldelade stänger kanalen med kommandot Channel.Close och klienten får ett undantag OperationInterruptedException, som kommer att innehålla felkoden 403 och dess beskrivning.

När kön har varit inaktiv i mer än 10 sekunder, går den in i viloläge, vilket anropar GC i kön och därmed avsevärt minskar minnesbehovet för denna kö.

Skapa en kö via användargränssnittet

Gå till administrationspanelen RabbitMQ under användaren guest (användarnamn: guest och lösenord: guest). Observera att användaren guest bara kan ansluta från localhost. Låt oss nu gå till fliken Köer och klicka på Lägg till en ny kö. Fyll i egenskaperna:

Så snart vi har angett alla nödvändiga uppgifter och klickar på Lägg till köer, kommer kön att visas i den övergripande listan.

Klicka på köens namn för att visa detaljerad information. Här kan du konfigurera bindningen mellan utbytet och kön, se listan consumers, publicera/mottaga meddelanden, ta bort kön och visa statistik.

Skapa Bindning

Att skapa en bindning görs genom en synkron RPC begäran till servern. Begäran görs genom metoden Queue.Bind, som anropas med följande parametrar:

Kö namn
namnet på utbytet
andra parametrar

Exempel på skapande av bindning via RabbitMQ.Client:

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

queue — namnet på kön
exchange — namnet på utbytet
routingKey — rutt nyckel
argument — valfria argument

Om skapandet av bindning är möjligt, så kommer servern att skicka ett synkront RPC svar Queue.BindOk.

Skapa Bindning via användargränssnittet

Så snart vi har angett alla nödvändiga uppgifter och klickar på Bind, kommer bindningen att visas i den övergripande listan:

Code

I detta avsnitt beskriver vi kön och bindningen med kod i C#, som om vi skulle utveckla ett bibliotek. Det kan vara användbart för förståelsen.

public interface IQueue
    {        
        string Name { get; }

        // 
        //     Om det sätts till true, kommer kön att vara permanent.
        //     Den kommer att lagras på disken och kan
        //     överleva omstart av servern/brokern.
        //     Om värdet är false, är kön temporär och kommer att tas bort,
        //     när servern/brokern startas om
        // 
        bool IsDurable { get; }

        // 
        //     Om värdet är true, kommer
        //     denna kö bara att tillåta anslutning
        //     av en konsument
        // 
        bool IsExclusive { get; }

        // 
        //     Automatisk borttagning.
        //     Kön kommer att tas bort när alla klienter kopplar ifrån.
        // 
        bool IsAutoDelete { get; }

        // 
        //     Valfria argument
        // 
        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} får inte vara 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ågprioriterat läge. Lågprioriterat läge tvingar att spara 
        //     så många meddelanden som möjligt på disk för att minska 
        //     användningen av minnet
        // 
        public const string Lazy = "lazy";
    }

public interface IBinding
    {
        // 
        //     Koppling som kommer att binda en förbindelse
        // 
        IExchange Exchange { get; }

        // 
        //     Routingnyckel
        // 
        string RoutingKey { get; }

        // 
        //     Valfria argument
        // 
        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; }
    }

Källa: habr.com