🥇RabbitMQ. Teil 3. Wir beschäftigen uns mit Queues und Bindings

Warteschlange (Warteschlange) — eine Datenstruktur auf der Festplatte oder im Arbeitsspeicher, die Verweis auf Nachrichten speichert und deren Kopien bereitstellt. Konsumenten (der Konsumenten). Warteschlange stellt dar einen Erlang-Prozess mit Zustand (in dem die Nachrichten auch zwischengespeichert werden können). 1.000 Warteschlangen können etwa 80 MB belegen.

Bindung (Bindung) — eine Regel, die dem Broker mitteilt, in welche Warteschlange die Nachrichten zugestellt werden sollen.

Inhaltsverzeichnis

RabbitMQ. Teil 1. Einführung. Erlang, AMQP und RPC
RabbitMQ. Teil 2. Verständnis der Exchanges
RabbitMQ. Teil 3. Verständnis von Queues und Bindings
RabbitMQ. Teil 4. Was sind Nachrichten und Frames?
RabbitMQ. Teil 5. Leistung bei der Veröffentlichung 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. Überwachung

Temporäre Warteschlangen

Wenn eine Warteschlange mit dem Parameter autoDeleteerstellt wird, dann erhält diese Warteschlange die Fähigkeit sich automatisch zu löschen. Solche Warteschlangen werden normalerweise beim Anschluss des ersten Clients erstellt und gelöscht, wenn alle Clients getrennt werden.

Wenn eine Warteschlange mit dem Parameter exclusive, dann erlaubt diese Warteschlange nur einem Konsumenten die Verbindung. und wird entfernt, wenn der Kanal geschlossen wird. Solange der Kanal nicht geschlossen ist, kann der Kunde sich ab- und wieder anmelden, jedoch nur innerhalb derselben Verbindung. Wenn der Parameter exclusive eingerichtet ist, hat der Parameter autoDelete keine Wirkung.

Besonderheiten:

Bei kurzzeitigem Verbindungsabbruch werden wir Nachrichten verlieren, die noch nicht beim Verbraucher angekommen sind.
man kann das Phänomen Binding Churn. Dieses Phänomen entsteht, wenn die Anzahl der Erstellungs- und Löschvorgänge von Warteschlangen und Bindungen sehr hoch wird. Im Clusterbetrieb wird dieser Strom von Operationen auf alle Knoten verteilt und erzeugt erhebliche Belastung. Dieser Prozess kann optimiert werden, indem die Anzahl der Abonnements kontrolliert wird.

Dauerhafte Warteschlangen

Wenn eine Warteschlange mit dem Parameter durable, dann erlaubt diese Warteschlange bewahrt ihren Zustand und wird nach dem Neustart des Servers/Brokers wiederhergestellt. Diese Warteschlange existiert, bis der Befehl Queue.Delete.

Hochverfügbare Warteschlangen

HA-Warteschlangen erfordern eine Cluster-Umgebung von RabbitMQ. Im Clusterbetrieb werden alle Informationen über Austausch, 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 zu dieser HA-Warteschlange gehört. Nachdem die Nachricht von einem der Knoten konsumiert wurde, werden alle Kopien dieser Nachricht auf den anderen Knoten gelöscht.

HA-Warteschlangen können sich über alle Knoten in einem Cluster erstrecken oder nur auf einzelne Knoten beschränkt sein.

Besonderheiten:

Die Nutzung von HA-Warteschlangen führt zu Leistungseinbußen. Wenn eine Nachricht in eine HA-Warteschlange eingefügt oder aus einer HA-Warteschlange konsumiert wird, muss RabbitMQ die Koordination zwischen allen Knoten durchführen (2-3 Knoten sind in der Regel ausreichend).

Warteschlange erstellen

Das Erstellen einer Warteschlange erfolgt über einen synchronen RPC Anruf an den Server. Der Aufruf erfolgt über die Methode Queue.Declare, die mit den folgenden Parametern aufgerufen wird:

Name der Warteschlange
weitere Parameter

Ein Beispiel für die 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 einem Systemnamen der Warteschlange übereinstimmen.
durable — Wenn true, wird die Warteschlange ihren Zustand speichern und nach einem Neustart des Servers/Brokers wiederhergestellt.
exclusive — wenn true, erlaubt die Warteschlange nur einem Verbraucher den Zugriff.
autoDelete — wenn true, erlangt die Warteschlange die Fähigkeit sich automatisch zu löschen
Argumente — optionale Argumente. Im Folgenden erläutern wir dies näher.

Argumente

x-message-ttl(x-message-time-to-live) — ermöglicht die Festlegung einer Verfallszeit für die Nachricht in Millisekunden. Wenn die Warteschlange mit einem festgelegten Argumentwert erstellt wird, x-message-ttlwird diese Warteschlange automatisch Nachrichten ausschließen, deren Ablaufdatum überschritten ist.Durch die Festlegung eines Argumentwerts x-message-ttl wird das maximale Alter für alle Nachrichten in dieser Warteschlange definiert. Das Erstellen einer solchen Warteschlange hilft, das Empfangen veralteter Informationen zu verhindern.Dies kann in Echtzeitsystemen verwendet werden. Wenn für eine Warteschlange, für die ein Exchange für abgelehnte Nachrichten festgelegt ist, ein Argumentwert definiert wird, x-message-ttlbeginnen die abgelehnten Nachrichten in dieser Warteschlange eine Lebensdauer zu haben..
x-expires — legt die Zeit in Millisekunden fest, nach deren Ablauf die Warteschlange gelöscht wird. Eine Warteschlange kann ihre Gültigkeit nur verlieren, wenn sie keine Abonnenten hat. Ist die Warteschlange mit Abonnenten verbunden, kann sie sich nur dann automatisch löschen, wenn alle Abonnenten" Basic.Cancel oder sich abmelden. Die Lebensdauer der Warteschlange kann nur enden, wenn keine Anfrage an sie gerichtet wurde, Basic.Get. Andernfalls wird der aktuelle Wert der Lebensdauer zurückgesetzt, und die Warteschlange wird nicht mehr automatisch gelöscht. Außerdem gibt es keine Garantie dafür, wie schnell die Löschung der Warteschlange nach Ablauf ihrer Lebensdauer erfolgt..
x-max-length — definiert die maximale Anzahl von Nachrichten in der Warteschlange. Wenn die Anzahl der Nachrichten in der Warteschlange die maximale Anzahl überschreitet, beginnen die ältesten Nachrichten gelöscht zu werden.

x-max-length-bytes — legt die maximal zulässige Gesamtdatenmenge der Nachrichten in der Warteschlange fest. Wenn der festgelegte Wert überschritten wird (eine Überlauf der Warteschlange tritt bei der nächsten Veröffentlichung einer Nachricht auf), werden die ältesten Nachrichten gelöscht.
x-overflow — Dieses Argument wird verwendet, um das Verhalten im Falle eines Warteschlangenüberlaufs zu steuern. 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 unterbrochen.
x-dead-letter-exchange — Legt die Exchange fest, in die abgelehnte Nachrichten gesendet werden, die nicht erneut in die Warteschlange eingereiht werden.
x-dead-letter-routing-key — Legt einen optionalen Routing-Schlüssel für abgelehnte Nachrichten fest.
x-max-priority — Ermöglicht die PrioritätsSortierung in der Warteschlange mit einem maximalen Prioritätswert von 255 (RabbitMQ-Versionen 3.5.0 und höher). Die Zahl gibt die maximale Priorität an, die die Warteschlange unterstützen wird. Wenn das Argument nicht gesetzt ist, unterstützt die Warteschlange keine Nachrichtenprioritäten.
x-queue-mode — Ermöglicht es, die Warteschlange in den schlummernden Moduszu versetzen. In diesem Modus werden so viele Nachrichten wie möglich auf der Festplatte gespeichert. Der Verbrauch von RAM wird minimal sein. Wenn es nicht gesetzt ist, speichert die Warteschlange Nachrichten im Speicher, um eine möglichst schnelle Lieferung zu gewährleisten.
x-queue-master-locator — wenn wir ein Cluster haben, kann eine Master-Warteschlange eingestellt werden
x-ha-policy — wird beim Erstellen von HA-Warteschlangen verwendet und bestimmt, wie die Nachricht über die Knoten verteilt wird. Wenn der Wert gesetzt ist all, wird die Nachricht auf allen Knoten gespeichert. Wenn der Wert gesetzt ist nodes, wird die Nachricht auf bestimmten Knoten des Clusters gespeichert
x-ha-nodes — definiert die Knoten, denen eine bestimmte Warteschlange zugeordnet ist HA

Wenn die Warteschlange erstellt wird möglich, sendet der Server eine synchrone RPC Antwort Queue.DeclareOk. Wenn die Warteschlange nicht erstellt werden kann (Fehler beim Anforderungsauftrag Queue.Declare), dann wird der Kanal vom Server mit dem Befehl Channel.Close geschlossen und der Client erhält eine Ausnahme OperationInterruptedException , die einen Fehlercode und eine Beschreibung enthalten wird.Ein erneuter Aufruf

mit ähnlichen Parametern Queue.Declare liefert nützliche Informationen über diese Warteschlange. Beispielsweise die Gesamtzahl der Nachrichten, die in dieser Warteschlange warten, und die Gesamtzahl der darauf registrierten Verbraucher. Ein Aufruf

Aufruf Queue.Declare unter Benutzeranmeldedaten, denen die erforderlichen Rechte nicht zugewiesen sind schließt den Kanal mithilfe des Befehls geschlossen und der Client erhält eine Ausnahme OperationInterruptedException , die einen Fehlercode und eine Beschreibung enthalten wird., der einen Fehlercode 403 und eine Beschreibung enthalten wird.

Nachdem die Warteschlange länger als 10 Sekunden inaktiv ist, geht sie in den Ruhezustand, was die GC in der Warteschlange auslöst und zu einer signifikanten Reduzierung des benötigten Speichers für diese Warteschlange führt.

Erstellen einer Warteschlange über die grafische Benutzeroberfläche

Loggen Sie sich im Administrationsbereich ein RabbitMQ mit dem Benutzer guest (Benutzername: guest und Passwort: guest). Beachten Sie, dass sich der Benutzer guest nur von localhost aus anmelden kann. Jetzt wechseln wir zum Tab Queues und klicken auf Eine neue Warteschlange hinzufügen. Wir füllen die Eigenschaften aus:

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

Ein Klick auf den Namen der Warteschlange zeigt ihre Details an. Hier können Sie die Bindung zwischen dem Exchange und der Warteschlange konfigurieren, die Liste Konsumenteneinsehen, Nachrichten veröffentlichen/erhalten, die Warteschlange löschen und Statistiken anzeigen.

Erstellen einer Bindung

Die Erstellung der Bindung erfolgt mithilfe von RPC Anruf an den Server. Der Aufruf erfolgt über die Methode Queue.Bind, die mit den folgenden Parametern aufgerufen wird:

Name der Warteschlange
Name des Exchange
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 — Name der Warteschlange
exchange — Name des Exchanges
routingKey — Schlüssel zur Weiterleitung
Argumente — optionale Argumente

Wenn die Bindung erstellt wird, möglich, sendet der Server eine synchrone RPC Antwort Queue.BindOk.

Erstellen einer Bindung über die grafische Benutzeroberfläche

Loggen Sie sich im Administrationsbereich ein RabbitMQ mit dem Benutzer guest (Benutzername: guest und Passwort: guest). Beachten Sie, dass sich der Benutzer guest nur von localhost aus anmelden kann. Jetzt wechseln wir zum Tab Queues und klicken Sie auf die Warteschlange my_queue. Füllen Sie die Felder im Abschnitt aus Bindings:

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

Code

In diesem Abschnitt beschreiben wir die Warteschlange und die Bindung mit C#, als ob wir eine Bibliothek entwickeln müssten. Dies könnte hilfreich sein, um das Konzept besser zu verstehen.

public interface IQueue
    {        
        string Name { get; }

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

        // 
        //     Wenn der Wert true ist, 
        //     erlaubt diese Warteschlange den Zugriff 
        //     nur für einen Verbraucher.
        // 
        bool IsExclusive { get; }

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

        // 
        //     Optionale Argumente
        // 
        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} darf nicht null sein");

            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";
        // 
        //     Fauler Modus. Der faule Modus verpflichtet dazu, 
        //     möglichst viele Nachrichten auf der Festplatte zu speichern, 
        //     um den Einsatz von RAM zu verringern
        // 
        public const string Lazy = "lazy";
    }

public interface IBinding
    {
        // 
        //     Der Austausch, der mit der Bindung verbunden ist
        // 
        IExchange Exchange { get; }

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

        // 
        //     Optionale Argumente
        // 
        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; }
    }

Quelle: habr.com