Muster „Konkurrierende Consumer“

Mehreren gleichzeitigen Consumern die Verarbeitung von Nachrichten ermöglichen, die auf dem gleichen Messagingkanal empfangen werden Durch mehrere gleichzeitige Consumer kann ein System mehrere Nachrichten parallel verarbeiten, um den Durchsatz zu optimieren, Skalierbarkeit und Verfügbarkeit zu erhöhen und die Workload auszugleichen.

Kontext und Problem

Eine in der Cloud ausgeführte Anwendung soll eine große Anzahl von Anforderungen verarbeiten. Anstatt jede Anforderung synchron zu verarbeiten, ist es ein gängiges Verfahren, die Anforderungen über ein Messagingsystem an einen anderen Dienst (einen Consumerdienst) zu übergeben, der sie asynchron verarbeitet. Diese Strategie trägt dazu bei, dass die Geschäftslogik in der Anwendung nicht blockiert wird, während Anforderungen verarbeitet werden.

Die Anzahl der Anforderungen kann im Lauf der Zeit aus verschiedenen Gründen stark abweichen. Ein plötzlicher Anstieg der Benutzeraktivität oder der aggregierten Anfragen von mehreren Mandanten kann zu einer nicht vorhersagbaren Workload führen. Zu Spitzenzeiten muss ein System möglicherweise viele Hunderte von Anforderungen pro Sekunde verarbeiten, während zu anderen Zeiten die Anzahl sehr klein werden kann. Darüber hinaus kann sich die Art der durchgeführten Arbeit für die Verarbeitung dieser Anforderungen stark unterscheiden. Wenn Sie eine einzelne Instanz des Consumerdiensts verwenden, können Sie dafür sorgen, dass diese Instanz mit Anforderungen überflutet wird. Oder das Messagingsystem kann durch einen Zustrom von Nachrichten aus der Anwendung überlastet werden. Zur Verarbeitung dieser schwankenden Workload kann das System mehrere Instanzen des Consumerdiensts ausführen. Diese Consumer müssen jedoch koordiniert werden, um sicherzustellen, dass jede Nachricht nur an einen einzelnen Consumer übermittelt wird. Außerdem muss für die Workload ein Lastenausgleich über alle Consumer stattfinden, um zu verhindern, dass eine Instanz zu einem Engpass wird.

Lösung

Implementieren Sie den Kommunikationskanal zwischen der Anwendung und den Instanzen des Consumerdiensts mithilfe einer Warteschlange. Die Anwendung sendet Anforderungen in Form von Nachrichten an die Warteschlange, und die Consumerdienstinstanzen empfangen Nachrichten aus der Warteschlange und verarbeiten diese. Durch diesen Ansatz kann derselbe Pool von Consumerdienstinstanzen Nachrichten von jeder Instanz der Anwendung verarbeiten. Die Abbildung veranschaulicht die Verwendung einer Nachrichtenwarteschlange zur Verteilung von Arbeit an Instanzen eines Diensts.

Diese Lösung hat folgende Vorteile:

• Sie bietet ein für unterschiedliche Lasten abgeglichenes System, das starke Schwankungen in der Menge der von Anwendungsinstanzen gesendeten Anfragen verarbeiten kann. Die Warteschlange fungiert als Puffer zwischen den Anwendungsinstanzen und den Consumerdienstinstanzen. Dieser Puffer kann dazu beitragen, die Auswirkungen auf die Verfügbarkeit und Reaktionsfähigkeit sowohl der Anwendung als auch der Dienstinstanzen zu minimieren. Weitere Informationen finden Sie unter Warteschlangenbasiertes Belastungsausgleichsmuster. Eine lang andauernde Verarbeitung einer Nachricht behindert nicht die gleichzeitige Verarbeitung anderer Nachrichten durch andere Instanzen des Consumerdiensts.

• Dadurch wird die Zuverlässigkeit verbessert. Wenn ein Produzent, anstatt dieses Muster zu verwenden, direkt mit einem Consumer kommuniziert, diesen jedoch nicht überwacht, besteht eine hohe Wahrscheinlichkeit, dass Nachrichten verloren gehen oder nicht verarbeitet werden, wenn beim Consumer ein Fehler auftritt. Bei diesem Muster werden Nachrichten nicht an eine bestimmte Dienstinstanz gesendet. Wenn bei einer Dienstinstanz ein Fehler auftritt, wird kein Produzent blockiert, und Nachrichten können von jeder funktionsfähigen Dienstinstanz verarbeitet werden.

• Es ist keine komplexe Koordination zwischen den Consumern oder zwischen dem Produzenten und den Consumerinstanzen erforderlich. Die Nachrichtenwarteschlange stellt sicher, dass jede Nachricht mindestens einmal übermittelt wird.

• Sie ist skalierbar. Bei Verwendung der automatischen Skalierung kann das System die Instanzenanzahl des Consumerdiensts dynamisch erhöhen oder verringern, wenn die Nachrichtenanzahl schwankt.

• Dies kann die Resilienz verbessern, wenn die Nachrichtenwarteschlange transaktionale Lesevorgänge bereitstellt. Wenn eine Consumerdienstinstanz die Nachricht als Teil eines Transaktionsvorgang liest und verarbeitet und bei der Consumerdienstinstanz ein Fehler auftritt, kann dieses Muster sicherstellen, dass die Nachricht an die Warteschlange zurückgegeben wird, wo sie von einer anderen Instanz des Consumerdiensts abgerufen und verarbeitet werden kann. Um das Risiko zu verringern, dass bei einer Nachricht kontinuierlich Fehler auftreten, empfiehlt sich die Nutzung von Warteschlangen für unzustellbare Nachrichten.

Probleme und Überlegungen

Beachten Sie die folgenden Punkte bei der Entscheidung, wie dieses Muster implementiert werden soll:

• Nachrichtensortierung: Die Reihenfolge, in der Consumerdienstinstanzen Nachrichten empfangen, ist nicht garantiert und entspricht nicht unbedingt der Reihenfolge, in der die Nachrichten erstellt wurden. Entwerfen Sie das System so, dass die Verarbeitung von Nachrichten mit Sicherheit idempotent ist, da auf diese Weise jede Abhängigkeit von der Reihenfolge der Nachrichtenverarbeitung vermieden werden kann. Weitere Informationen finden Sie unter Idempotenzmuster im Blog von Jonathan Oliver.

  Microsoft Azure Service Bus-Warteschlangen können eine garantierte FIFO-Sortierung (First In, First Out) von Nachrichten mithilfe von Nachrichtensitzungen implementieren. Weitere Informationen finden Sie unter Messagingmuster mithilfe von Sitzungen.

• Entwerfen von Diensten für Resilienz: Wenn das System für das Erkennen und Neustarten von fehlerhaften Dienstinstanzen ausgelegt ist, kann es erforderlich sein, die von den Dienstinstanzen durchgeführte Verarbeitung als idempotente Vorgänge zu implementieren, damit es minimale Auswirkungen hat, wenn eine einzelne Nachricht mehrmals abgerufen und verarbeitet wird.

• Erkennen von nicht verarbeitbaren Nachrichten: Eine falsch formatierte Nachricht oder eine Aufgabe, die Zugriff auf nicht verfügbare Ressourcen benötigt, kann zu Fehlern in einer Dienstinstanz führen. Das System sollte verhindern, dass solche Nachrichten an die Warteschlange zurückgegeben werden, und stattdessen die Details dieser Nachrichten an anderer Stelle erfassen und speichern, damit sie bei Bedarf analysiert werden können.

• Verarbeiten von Ergebnissen: Die Dienstinstanz, die eine Nachricht verarbeitet, ist vollständig von der Anwendungslogik, die die Nachricht generiert, entkoppelt und kann möglicherweise nicht direkt mit dieser kommunizieren. Wenn die Dienstinstanz Ergebnisse generiert, die an die Anwendungslogik zurückgegeben werden müssen, müssen diese Informationen an einem Ort gespeichert werden, der für beide zugänglich ist. Um zu verhindern, dass die Anwendungslogik unvollständige Daten abruft, muss das System angeben, wann die Verarbeitung abgeschlossen ist.

  Wenn Sie Azure verwenden, kann ein Workerprozess mithilfe einer dedizierten Nachrichtenantwort-Warteschlange Ergebnisse zurück an die Anwendungslogik übergeben. Die Anwendungslogik muss in der Lage sein, diese Ergebnisse mit der ursprünglichen Nachricht zu korrelieren. Dieses Szenario wird unter Einführung in asynchrone Nachrichten ausführlicher beschrieben.

• Skalieren des Messagingsystems: In einer groß angelegten Lösung könnte eine einzelne Warteschlange durch die Anzahl von Nachrichten überlastet werden und einen Engpass im System darstellen. Ziehen Sie in diesem Fall in Betracht, das Nachrichtensystem zu partitionieren, sodass Nachrichten von bestimmten Produzenten an eine bestimmte Warteschlange gesendet werden, oder verwenden Sie einen Lastenausgleich, um Nachrichten auf mehrere Warteschlangen zu verteilen.

• Sicherstellen der Zuverlässigkeit des Messagingsystems: Ein zuverlässiges Messagingsystem muss gewährleisten, dass eine Nachricht nach dem Einstellen in eine Warteschlange durch eine Anwendung nicht verloren geht. Dieses System ist besonders wichtig, um sicherzustellen, dass alle Nachrichten mindestens einmal übermittelt werden.

Verwendung dieses Musters

Verwenden Sie dieses Muster in folgenden Fällen:

• Die Workload für eine Anwendung wird in Aufgaben unterteilt, die asynchron ausgeführt werden können.
• Aufgaben sind unabhängig voneinander und können parallel ausgeführt werden.
• Die Menge der Arbeit kann stark schwanken, sodass eine skalierbare Lösung erforderlich ist.
• Die Lösung muss Hochverfügbarkeit bereitstellen und stabil bleiben, wenn bei der Verarbeitung einer Aufgabe ein Fehler auftritt.

Dieses Muster ist in folgenden Fällen möglicherweise nicht geeignet:

• Es ist nicht einfach, die Anwendungsworkload in einzelne Aufgaben zu trennen, oder es besteht eine starke Abhängigkeit zwischen Aufgaben.
• Aufgaben müssen synchron ausgeführt werden, und die Anwendungslogik muss vor dem Fortfahren warten, bis eine Aufgabe abgeschlossen ist.
• Aufgaben müssen in einer bestimmten Reihenfolge ausgeführt werden.

Einige Messagingsysteme unterstützen Sitzungen, mit denen ein Produzent Nachrichten gruppieren und sicherstellen kann, dass sie alle vom selben Consumer verarbeitet werden. Dieser Mechanismus kann mit priorisierte Nachrichten verwendet werden (sofern diese unterstützt werden), um eine Form der Nachrichtensortierung zu implementieren, bei der Nachrichten von einem Produzenten an einen einzelnen Consumer nacheinander übermittelt werden.

Workloadentwurf

Ein Architekt sollte evaluieren, wie das Muster „Competing Consumers“ im Design seines Workloads verwendet werden kann, um die Ziele und Prinzipien zu erreichen, die in den Säulen des Azure Well-Architected Framework behandelt werden. Zum Beispiel:

Berücksichtigen Sie wie bei jeder Designentscheidung alle Kompromisse im Hinblick auf die Ziele der anderen Säulen, die mit diesem Muster eingeführt werden könnten.

Beispiel

Azure bietet Service Bus-Warteschlangen und Warteschlangentrigger für Azure-Funktionen, die in Kombination eine direkte Implementierung dieses Cloudentwurfsmusters darstellen. Azure-Funktionen werden über Trigger und Bindungen in Azure Service Bus integriert. Die Service Bus-Integration ermöglicht die Erstellung von Funktionen zur Nutzung von Warteschlangennachrichten, die von Veröffentlichern gesendet werden. Von den veröffentlichenden Anwendungen werden Nachrichten an eine Warteschlange gesendet, und als Azure-Funktionen implementierte Consumer können Nachrichten aus dieser Warteschlange abrufen und verarbeiten.

Aus Resilienzgründen ermöglicht eine Service Bus-Warteschlange einem Consumer beim Abrufen einer Nachricht aus der Warteschlange die Verwendung des Modus PeekLock. In diesem Modus wird die Nachricht nicht entfernt, sondern lediglich für andere Consumer unsichtbar gemacht. Von der Azure Functions-Runtime wird eine Nachricht im PeekLock-Modus empfangen. Nach erfolgreicher Beendigung der Funktion wird „Complete“ (Abschließen) für die Nachricht aufgerufen. Wurde die Funktion nicht erfolgreich abgeschlossen, wird ggf. „Abandon“ (Abbrechen) aufgerufen, und die Nachricht wird wieder sichtbar, sodass sie von einem anderen Consumer abgerufen werden kann. Wenn die Ausführungsdauer der Funktion das PeekLock-Timeout übersteigt, wird die Sperre automatisch verlängert, solange die Funktion ausgeführt wird.

Azure-Funktionen können auf der Grundlage der Warteschlangentiefe skaliert werden und agieren jeweils als konkurrierende Consumer der Warteschlange. Werden mehrere Instanzen der Funktionen erstellt, konkurrieren sie jeweils miteinander, indem Sie die Nachrichten unabhängig voneinander abrufen und verarbeiten.

Ausführliche Informationen zur Verwendung von Azure Service Bus-Warteschlangen finden Sie unter Service Bus-Warteschlangen, -Themen und -Abonnements.

Informationen zu Azure-Funktionen mit Warteschlangentrigger finden Sie unter Azure Service Bus-Trigger für Azure Functions.

Der folgende Code zeigt, wie Sie eine neue Nachricht erstellen und unter Verwendung einer Instanz vom Typ ServiceBusClient an eine Service Bus-Warteschlange senden können:

private string serviceBusConnectionString = ...;
...

  public async Task SendMessagesAsync(CancellationToken  ct)
  {
   try
   {
    var msgNumber = 0;

    var serviceBusClient = new ServiceBusClient(serviceBusConnectionString);

    // create the sender
    ServiceBusSender sender = serviceBusClient.CreateSender("myqueue");

    while (!ct.IsCancellationRequested)
    {
     // Create a new message to send to the queue
     string messageBody = $"Message {msgNumber}";
     var message = new ServiceBusMessage(messageBody);

     // Write the body of the message to the console
     this._logger.LogInformation($"Sending message: {messageBody}");

     // Send the message to the queue
     await sender.SendMessageAsync(message);

     this._logger.LogInformation("Message successfully sent.");
     msgNumber++;
    }
   }
   catch (Exception exception)
   {
    this._logger.LogException(exception.Message);
   }
  }

Das folgende Codebeispiel zeigt einen als C#-Azure-Funktion geschriebenen Consumer, von dem Nachrichtenmetadaten gelesen werden und eine Service Bus-Warteschlangennachricht protokolliert wird. Beachten Sie die Verwendung des Attributs ServiceBusTrigger zur Bindung an eine Service Bus-Warteschlange.

[FunctionName("ProcessQueueMessage")]
public static void Run(
    [ServiceBusTrigger("myqueue", Connection = "ServiceBusConnectionString")]
    string myQueueItem,
    Int32 deliveryCount,
    DateTime enqueuedTimeUtc,
    string messageId,
    ILogger log)
{
    log.LogInformation($"C# ServiceBus queue trigger function consumed message: {myQueueItem}");
    log.LogInformation($"EnqueuedTimeUtc={enqueuedTimeUtc}");
    log.LogInformation($"DeliveryCount={deliveryCount}");
    log.LogInformation($"MessageId={messageId}");
}

Nächste Schritte

• Einführung in asynchrone Nachrichten. Warteschlangen sind ein Mechanismus für asynchrone Kommunikation. Wenn ein Consumerdienst eine Antwort an eine Anwendung senden muss, kann es erforderlich sein, ein Antwortmessaging zu implementieren. Die Einführung in asynchrone Nachrichten enthält Informationen zum Implementieren des Anforderung-Antwort-Messaging mithilfe von Nachrichtenwarteschlangen.

• Leitfaden für die automatische Skalierung. Es kann möglich sein, Instanzen eines Consumerdiensts zu starten und zu beenden, da die Länge der Warteschlange, in die Anwendungen Nachrichten senden, variiert. Automatische Skalierung kann dabei helfen, den Durchsatz während Zeiten hoher Verarbeitung beizubehalten.

Zugehörige Ressourcen

Die folgenden Muster und Anweisungen könnten für die Implementierung dieses Musters relevant sein:

• Muster „Computeressourcenkonsolidierung“: Eventuell ist es möglich, mehrere Instanzen eines Consumerdiensts in einem einzelnen Prozess zusammenzuführen, um Kosten und Verwaltungsaufwand zu reduzieren. Das Computeressourcen-Konsolidierungsmuster beschreibt die Vor- und Nachteile dieses Ansatzes.

• Muster „Warteschlangenbasierter Lastenausgleich“: Durch die Einführung einer Nachrichtenwarteschlange kann eine höhere Resilienz des Systems erreicht werden, sodass Dienstinstanzen stark abweichende Mengen von Anforderungen von Anwendungsinstanzen verarbeiten können. Die Nachrichtenwarteschlange fungiert als Puffer, in dem die Last ausgeglichen wird. Das warteschlangenbasierte Lastenausgleichsmuster beschreibt dieses Szenario ausführlicher.