🥇RabbitMQ. Teil 3. Wir befassen uns mit Warteschlangen und Bindungen

Warteschlange (Warteschlange) — eine Datenstruktur auf der Festplatte oder im Arbeitsspeicher, die Verweise auf Nachrichten speichert und deren Kopien übergibt consumers (Verbraucher). Warteschlange stellt dar Erlang-Prozess mit Zustand (in dem auch die Nachrichten zwischengespeichert werden können). 1.000 Warteschlangen können etwa 80 Mb in Anspruch nehmen.

Bindung (Binding) — eine Regel, die dem Exchange mitteilt, in welche Warteschlangen die Nachrichten gelangen sollen.

Inhaltsverzeichnis

RabbitMQ. Teil 1. Einführung. Erlang, AMQP und RPC
RabbitMQ. Teil 2. Verstehen der Exchanges
RabbitMQ. Teil 3. Wir beschäftigen uns mit Queues und Bindings
RabbitMQ. Teil 4. Wir klären, was Nachrichten und Frames sind
RabbitMQ. Teil 5. Leistung bei der Publikation und dem Konsum von Nachrichten
RabbitMQ. Teil 6. Überblick über die Module Federation und Shovel
RabbitMQ. Teil 7. Detaillierte Informationen zu Verbindung und Kanal
RabbitMQ. Teil 8. RabbitMQ in .NET
RabbitMQ. Teil 9. Monitoring

Temporäre Warteschlangen

Wenn die Warteschlange mit dem gesetzten Parameter erstellt wird autoDelete, dann erhält diese Warteschlange die Fähigkeit sich selbst automatisch zu löschen. Solche Warteschlangen werden in der Regel beim Anschluss des ersten Clients erstellt und gelöscht, wenn alle Clients getrennt sind.

Wenn die Warteschlange mit dem gesetzten Parameter erstellt wird exklusiv, dann erlaubt diese Warteschlange nur einem Verbraucher, sich zu verbinden und wird gelöscht, wenn der Kanal geschlossen wird. Solange der Kanal nicht geschlossen ist, kann der Client sich während der gleichen Verbindung trennen/verbinden. Wenn der Parameter exklusiv gesetzt ist, hat der Parameter autoDelete keine Auswirkungen.

Besonderheiten:

Bei kurzfristigen Verbindungsunterbrechungen verlieren wir Nachrichten, die noch nicht den Verbraucher erreicht haben.
Man kann das Phänomen Binding Churnbeobachten. Das Phänomen tritt auf, wenn die Anzahl der Operationen zum Erstellen/Löschen von Warteschlangen und Bindungen sehr hohe Werte erreicht. Im Clusterbetrieb wird dieser Umfang an Operationen sich über alle Knoten ausbreiten und eine erhebliche Belastung verursachen. Dieser Prozess kann durch die Kontrolle der Anzahl der Abonnements optimiert werden.

Dauerhafte Warteschlangen

Wenn die Warteschlange mit dem gesetzten Parameter erstellt wird durable, dann erlaubt diese Warteschlange bewahren ihren Zustand und stellen sich nach dem Neustart des Servers/Brokers wieder her. Diese Warteschlange wird existieren, bis der Befehl Queue.Delete.

Hochverfügbare Warteschlangen

HA Warteschlangen erfordern eine Clusterumgebung von RabbitMQ. Im Clustermodus werden alle Informationen über Exchanges, Warteschlangen, Bindungen und Verbraucher auf alle Knoten kopiert.

Wenn eine Nachricht in eine HA Warteschlange veröffentlicht wird, wird sie auf jedem Knoten gespeichert, der zur HA Warteschlange gehört. Sobald die Nachricht auf einem der Knoten konsumiert wird, werden alle Kopien dieser Nachricht auf den anderen Knoten gelöscht.

HA Warteschlangen können sich über alle Knoten in einem bestimmten Cluster oder nur über einzelne erstrecken.

Besonderheiten:

Die Verwendung von HA-Warteschlangen führt zu Leistungseinbußen. Beim Hinzufügen einer Nachricht zu einer HA-Warteschlange oder beim Abrufen einer Nachricht aus einer HA-Warteschlange muss RabbitMQ die Koordination über alle Knoten hinweg durchführen (in der Regel sind 2-3 Knoten ausreichend).

Warteschlange erstellen

Die Erstellung einer Warteschlange erfolgt synchron. RPC Anruf an den Server. Die Anfrage erfolgt mit der Methode Queue.Declare, die mit den Parametern aufgerufen wird:

Name der Warteschlange
weitere Parameter

Beispiel zur Erstellung einer Warteschlange mit RabbitMQ.Client:

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

queue — der Name der Warteschlange, die wir erstellen möchten. Der Name muss eindeutig sein und darf nicht mit dem Systemnamen der Warteschlange übereinstimmen.
durable — wenn true, speichert die Warteschlange ihren Zustand und stellt ihn nach einem Neustart des Servers/Brokers wieder her. — wenn true, erlaubt die Warteschlange nur einem Verbraucher den Zugriff.
exklusiv — wenn true, erhält die Warteschlange die Fähigkeit
autoDelete — optionale Argumente. Wir werden diese unten näher erläutern. sich selbst automatisch zu löschen
arguments x-message-ttl

arguments

x-message-time-to-live() — ermöglicht es, die Lebensdauer von Nachrichten in Millisekunden festzulegen. Wenn die Warteschlange mit einem festgelegten Wert für das Argument erstellt wird,wird diese Warteschlange x-message-time-to-liveautomatisch Nachrichten ausschließen, deren Gültigkeit abgelaufen ist. Das Setzen des Argumentwertslegt das maximale Alter für alle Nachrichten in dieser Warteschlange fest. Die Erstellung einer solchen Warteschlange x-message-time-to-live hilft, den Erhalt veralteter Informationen zu verhindern. Dies kann in Echtzeitsystemen verwendet werden. Wenn für die Warteschlange, der ein Austausch für abgelehnte Nachrichten zugewiesen ist, ein Wert für das Argument festgelegt wird,beginnen abgelehnte Nachrichten in dieser Warteschlange x-message-time-to-liveeine Lebensdauer zu haben. x-expires.
— legt den Wert in Millisekunden fest, nach dessen Ablauf die Warteschlange gelöscht wird. Die Warteschlange kann ihre Lebensdauer nur verlieren, wenn sie keine Abonnenten hat. Wenn Abonnenten mit der Warteschlange verbunden sind, kann sie sich nur dann automatisch löschen, wenn alle Abonnenten Basic.Cancel aufrufen oder sich abmelden. Die Lebensdauer der Warteschlange kann nur ablaufen, wenn keine Anforderung an sie gestellt wurde Basic.Get. Andernfalls wird der aktuelle Wert der Lebensdauer auf null gesetzt, und die Warteschlange wird nicht mehr automatisch gelöscht. Zudemgibt es keine Garantie dafür, wie schnell die Löschung der Warteschlange nach Ablauf ihrer Lebensdauer erfolgt. x-max-length.
x-max-length setzt die maximale Anzahl von Nachrichten in der Warteschlange fest. Wenn die Anzahl der Nachrichten in der Warteschlange die maximale Zahl überschreitet, werden die ältesten Nachrichten gelöscht.

x-max-lenght-bytes setzt die maximal zulässige Gesamtgröße der Nutzlastnachrichten in der Warteschlange fest. Bei Überschreitung des festgelegten Wertes (Warteschlangenüberlauf beim Veröffentlichen einer Nachricht) beginnen die ältesten Nachrichten gelöscht zu werden.
x-overflow dieses Argument wird verwendet, um das Verhalten im Fall eines Warteschlangenüberlaufs zu konfigurieren. Es sind zwei Werte verfügbar: drop-head (Standardwert) und reject-publish. Wenn Sie wählen, drop-head, werden die ältesten Nachrichten gelöscht. Wenn Sie wählen, reject-publish, wird der Empfang von Nachrichten ausgesetzt.
x-dead-letter-exchange setzt den Exchange fest, an den abgelehnte Nachrichten weitergeleitet werden, die nicht wieder in die Warteschlange gestellt werden.
x-dead-letter-routing-key setzt einen optionalen Routing-Schlüssel für abgelehnte Nachrichten fest.
x-max-priority ermöglicht die Sortierung nach Prioritäten in der Warteschlange mit einem maximalen Prioritätswert von 255 (RabbitMQ-Versionen 3.5.0 und höher). Diese Zahl gibt die maximale Priorität an, die die Warteschlange unterstützen wird. Wenn das Argument nicht festgelegt ist, unterstützt die Warteschlange keine Nachrichtenprioritäten.
x-queue-mode ermöglicht es, die Warteschlange in den faulen Moduszu versetzen. In diesem Modus werden so viele Nachrichten wie möglich auf der Festplatte gespeichert. Der Arbeitsspeicherverbrauch wird minimal sein. Wenn es nicht festgelegt ist, speichert die Warteschlange die Nachrichten im Arbeitsspeicher, um Nachrichten so schnell wie möglich zuzustellen.
x-queue-master-locator wenn wir einen Cluster haben, kann die Master-Warteschlange festgelegt werden.
x-ha-policy wird bei der Erstellung von HA-Warteschlangen verwendet und bestimmt, wie die Nachricht über die Knoten verteilt wird. Wenn der Wert festgelegt ist, all, wird die Nachricht auf allen Knoten gespeichert. Wenn der Wert festgelegt ist, nodes, wird die Nachricht auf bestimmten Knoten im Cluster gespeichert.
x-ha-nodes legt die Knoten fest, zu denen eine bestimmte Warteschlange gehören wird. HA

Wenn die Warteschlange erstellt wird, es könnte, dann wird der Server dem Client eine synchrone RPC Antwort Queue.DeclareOk. Wenn die Warteschlange vom Server mit dem Befehl erstellt wird, nicht möglich ist (Die Anfrage wurde abgelehnt Queue.Declare), dann wird der Kanal vom Server mit einem asynchronen Befehl ein erneuter Aufruf geschlossen und der Client erhält eine Ausnahme OperationInterruptedException, die den Fehlercode und die Beschreibung enthält.

mit denselben Parametern Queue.Declare liefert nützliche Informationen über diese Warteschlange. Zum Beispiel die Gesamtzahl der Nachrichten, die in dieser Warteschlange warten, und die Gesamtzahl der Abonnenten. gibt nützliche Informationen über diese Warteschlange zurück. Zum Beispiel die Gesamtzahl der Nachrichten, die in dieser Warteschlange warten, und die Gesamtzahl der darauf registrierten Verbraucher.

Aufruf Queue.Declare unter den Benutzerdaten, denen die erforderlichen Rechte nicht zugewiesen sind schließt den Kanal mit dem Befehl geschlossen und der Client erhält eine Ausnahme OperationInterruptedException, der den Fehlercode enthalten wird 403 und dessen Beschreibung.

Nachdem die Warteschlange für >= 10 Sekunden inaktiv war, geht sie in den Ruhezustand, indem sie GC in der Warteschlange aufruft, was zu einer erheblichen Reduzierung des Speichers führt, der für diese Warteschlange erforderlich ist.

Erstellen einer Warteschlange über die grafische Benutzeroberfläche

Melden Sie sich beim Administrationspanel RabbitMQ als Benutzer guest (Benutzername: guest und Passwort: guest). Beachten Sie, dass der Benutzer guest nur von localhost aus eine Verbindung herstellen kann. Lassen Sie uns nun zur Registerkarte Warteschlangen gehen und auf Eine neue Warteschlange hinzufügen. Eigenschaften ausfüllen:

Sobald wir alle erforderlichen Daten eingegeben haben und auf Warteschlangen hinzufügen, erscheint die Warteschlange in der Gesamtliste.

Ein Klick auf den Namen der Warteschlange zeigt ihre detaillierten Informationen. Hier können Sie die Bindung zwischen dem Exchange und der Warteschlange konfigurieren, die Liste anzeigen consumers, Nachrichten veröffentlichen/empfangen, die Warteschlange löschen und Statistiken einsehen.

Bindung erstellen

Die Bindung erfolgt durch eine synchrone RPC Anruf an den Server. Die Anfrage erfolgt mit der Methode Queue.Bind, die mit den Parametern aufgerufen wird:

Name der Warteschlange
Name des Exchanges
weitere Parameter

Beispiel für die Erstellung einer Bindung mit RabbitMQ.Client:

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

queue — Warteschlangenname
exchange — Name des Brokers
routingKey — Routing-Schlüssel
arguments — optionale Argumente

Wenn die Bindung erstellt wird es könnte, dann wird der Server dem Client eine synchrone RPC Antwort Queue.BindOk.

Erstellen einer Bindung über die grafische Benutzeroberfläche

Melden Sie sich beim Administrationspanel RabbitMQ als Benutzer guest (Benutzername: guest und Passwort: guest). Beachten Sie, dass der Benutzer guest nur von localhost aus eine Verbindung herstellen kann. Lassen Sie uns nun zur Registerkarte Warteschlangen und klicken Sie auf die Warteschlange my_queue. Füllen Sie die Felder im Abschnitt aus Bindings:

Sobald wir alle erforderlichen Daten eingegeben haben und auf Binden, wird die Bindung in der Gesamtliste angezeigt:

Code

In diesem Abschnitt beschreiben wir die Warteschlange und die Bindung mit C#-Code, als ob wir eine Bibliothek entwickeln müssten. Das könnte für das Verständnis hilfreich sein.

public interface IQueue
    {        
        string Name { get; }

        // 
        //     Wenn Sie true setzen, wird die Warteschlange dauerhaft sein. 
        //     Sie wird auf der Festplatte gespeichert und kann 
        //     einen Server-/Brokerneustart überstehen. 
        //     Wenn false, ist die Warteschlange temporär und wird gelöscht, 
        //     wenn der Server/Broker neu gestartet wird
        // 
        bool IsDurable { get; }

        // 
        //     Wenn der Wert true ist, 
        //     kann nur ein Consumer an diese Warteschlange angeschlossen werden
        // 
        bool IsExclusive { get; }

        // 
        //     Automatische Löschung. 
        //     Die Warteschlange wird gelöscht, wenn alle Clients getrennt sind.
        // 
        bool IsAutoDelete { get; }

        // 
        //     Optionale Argumente
        // 
        IDictionary Arguments { get; }
    }

öffentliche Klasse Warteschlange : IWarteschlange
    {
        öffentliche Warteschlange(
             string name, 
             bool istDauerhaft = true, 
             bool istExklusiv = false, 
             bool istAutomatischLöschen = false, 
             IDictionary argumente = null)
        {
            Name = name ??
                werfen eine ArgumentNullAusnahme(name, $"{name} darf nicht null sein");

            IstDauerhaft = istDauerhaft;
            IstExklusiv = istExklusiv;
            IstAutomatischLöschen = istAutomatischLöschen;
            Argumente = argumente ?? new Dictionary();
        }

        öffentliche string Name { get; }
        öffentliche bool IstDauerhaft { get; }
        öffentliche bool IstExklusiv { get; }
        öffentliche bool IstAutomatischLöschen { get; }
        öffentliche IDictionary Argumente { get; }
    }

öffentliche statische Klasse Wartemodus
    {       
        öffentliche konst string Standard = "default";
        // <summary>
        //    Fauler Modus. Der faule Modus wird dazu zwingen,
        //    so viele Nachrichten wie möglich auf der Festplatte zu speichern,
        //    um den Speicherverbrauch zu reduzieren
        // </summary>
        öffentliche konst string Faul = "lazy";
    }

öffentliches Interface IBinding
    {
        // <summary>
        //    Austausch, der durch Bindung verbunden wird
        // </summary>
        IExchange Exchange { get; }

        // <summary>
        //    Routing-Schlüssel
        // </summary>
        string RoutingKey { get; }

        // <summary>
        //    Optionale Argumente
        // </summary>
        IDictionary Argumente { get; }
    }

öffentliche Klasse Bindung : IBinding
    {
        öffentliche Bindung(
             IExchange exchange, 
             string routingKey, 
             IDictionary argumente)
        {
            Exchange = exchange;
            RoutingKey = routingKey;
            Argumente = argumente;
        }

        öffentliche IExchange Exchange { get; }
        öffentliche string RoutingKey { get; }
        öffentliche IDictionary Argumente { get; }
    }

Quelle: habr.com