Nachrichten in eine Reihenfolge bringen

Die Nachrichtenreihenfolge ist ein Feature in Pub/Sub, mit dem Sie Nachrichten in Ihren Abonnentenclients in der Reihenfolge empfangen können, in der sie von den Publisher-Clients veröffentlicht wurden.

Angenommen, ein Publisher-Client in einer Region veröffentlicht die Nachrichten 1, 2 und 3 in dieser Reihenfolge. Bei der Nachrichtenreihenfolge empfängt der Abonnentenclient die veröffentlichten Nachrichten in derselben Reihenfolge. Damit die Nachrichten in der richtigen Reihenfolge zugestellt werden, muss der Publisher-Client sie in derselben Region veröffentlichen. Abonnenten können sich jedoch mit jeder Region verbinden und die Reihenfolge wird weiterhin beibehalten.

Die Nachrichtenreihenfolge ist eine nützliche Funktion für Szenarien wie die Erfassung von Datenbankänderungen, die Verfolgung von Nutzersitzungen und Streaminganwendungen, bei denen die Chronologie von Ereignissen wichtig ist.

Auf dieser Seite wird das Konzept der Nachrichtenreihenfolge erläutert und beschrieben, wie Sie Ihre Abonnentenclients so einrichten, dass sie Nachrichten in der richtigen Reihenfolge empfangen. Informationen zum Konfigurieren Ihrer Publisher-Clients für die Nachrichtenreihenfolge finden Sie unter Sortierungsschlüssel verwenden, um eine Nachricht zu veröffentlichen.

Übersicht über die Nachrichtenreihenfolge

Die Reihenfolge in Pub/Sub wird durch Folgendes bestimmt:

• Sortierschlüssel: Dies ist ein String, der in den Pub/Sub-Nachrichtenmetadaten verwendet wird und die Entität darstellt, für die Nachrichten sortiert werden müssen. Der Sortierschlüssel kann bis zu 1 KB lang sein. Wenn Sie eine Reihe von sortierten Nachrichten in einer Region empfangen möchten, müssen Sie alle Nachrichten mit demselben Sortierungsschlüssel in derselben Region veröffentlichen. Beispiele für Sortierschlüssel sind Kundennummern und der Primärschlüssel einer Zeile in einer Datenbank.

  Der Veröffentlichungsdurchsatz für jeden Reihenfolgeschlüssel ist auf 1 MB/s begrenzt. Der Durchsatz für alle Sortierschlüssel für ein Thema ist auf das Kontingent begrenzt, das in einer Veröffentlichungsregion verfügbar ist. Dieses Limit kann auf viele GBps erhöht werden.

  Ein Reihenfolgeschlüssel ist nicht mit einer Partition in einem partitionsbasierten Messagingsystem vergleichbar, da Reihenfolgeschlüssel eine viel höhere Kardinalität als Partitionen haben.

• Nachrichtenreihenfolge aktivieren: Dies ist eine Aboeinstellung. Wenn für ein Abo die Nachrichtenreihenfolge aktiviert ist, empfangen die Abonnentenclients Nachrichten, die in derselben Region mit demselben Reihenfolgeschlüssel veröffentlicht wurden, in der Reihenfolge, in der sie vom Dienst empfangen wurden. Sie müssen diese Einstellung im Abo aktivieren.

  Angenommen, Sie haben zwei Abos A und B, die mit demselben Thema T verknüpft sind. Für Abo A ist die Nachrichtenreihenfolge aktiviert, für Abo B nicht. In dieser Architektur erhalten sowohl Abo A als auch Abo B dieselben Nachrichten aus dem Thema T. Wenn Sie Nachrichten mit Reihenfolgeschlüsseln in derselben Region veröffentlichen, empfängt Abo A die Nachrichten in der Reihenfolge ihrer Veröffentlichung. Bei Abo B werden die Nachrichten ohne erwartete Reihenfolge empfangen.

Wenn für Ihre Lösung Publisher-Clients sowohl geordnete als auch ungeordnete Nachrichten senden müssen, sollten Sie separate Themen erstellen, eines für geordnete und eines für ungeordnete Nachrichten.

Überlegungen bei der Verwendung von geordneten Nachrichten

Die folgende Liste enthält wichtige Informationen zum Verhalten von geordneten Nachrichten in Pub/Sub:

• Sortierung innerhalb des Schlüssels: Nachrichten, die mit demselben Sortierungsschlüssel veröffentlicht werden, sollten in der richtigen Reihenfolge empfangen werden. Angenommen, Sie veröffentlichen für den Sortierungsschlüssel A die Nachrichten 1, 2 und 3. Wenn die Sortierung aktiviert ist, wird erwartet, dass 1 vor 2 und 2 vor 3 geliefert wird.

• Schlüsselübergreifende Sortierung: Nachrichten, die mit unterschiedlichen Reihenfolgeschlüsseln veröffentlicht werden, werden nicht in der Reihenfolge ihres Erscheinens empfangen. Angenommen, Sie haben die Bestellschlüssel A und B. Für den Sortierschlüssel A werden die Nachrichten 1 und 2 in der richtigen Reihenfolge veröffentlicht. Für den Reihenfolgenschlüssel B werden die Nachrichten 3 und 4 in der richtigen Reihenfolge veröffentlicht. Nachricht 1 kann jedoch vor oder nach Nachricht 4 eintreffen.

• Wiederholte Nachrichtenzustellung: Pub/Sub stellt jede Nachricht mindestens einmal zu. Der Pub/Sub-Dienst kann Nachrichten also wiederholt zustellen. Wenn eine Nachricht noch einmal gesendet wird, werden auch alle nachfolgenden Nachrichten für diesen Schlüssel noch einmal gesendet, auch bestätigte. Angenommen, ein Abonnentenclient empfängt die Nachrichten 1, 2 und 3 für einen bestimmten Reihenfolgenschlüssel. Wenn Nachricht 2 noch einmal gesendet wird (weil die Bestätigungsfrist abgelaufen ist oder die Best-Effort-Bestätigung in Pub/Sub nicht beibehalten wurde), wird auch Nachricht 3 noch einmal gesendet. Wenn sowohl die Nachrichtenreihenfolge als auch ein Thema für unzustellbare Nachrichten für ein Abo aktiviert sind, ist dies möglicherweise nicht der Fall, da Pub/Sub Nachrichten an Themen für unzustellbare Nachrichten auf Best-Effort-Basis weiterleitet.

• Bestätigungsverzögerungen und Themen für unzustellbare Nachrichten: Nicht bestätigte Nachrichten für einen bestimmten Sortierschlüssel können die Zustellung von Nachrichten für andere Sortierschlüssel verzögern, insbesondere bei Serverneustarts oder Änderungen des Traffics. Um bei solchen Ereignissen Ordnung zu halten, sollten Sie alle Nachrichten rechtzeitig bestätigen. Wenn eine zeitnahe Bestätigung nicht möglich ist, sollten Sie ein Thema für unzustellbare Nachrichten verwenden, um zu verhindern, dass Nachrichten unbegrenzt lange aufbewahrt werden. Beachten Sie, dass die Reihenfolge möglicherweise nicht beibehalten wird, wenn Nachrichten in ein Thema für unzustellbare Nachrichten geschrieben werden.

• Nachrichtenaffinität (streamingPull-Clients): Nachrichten für denselben Schlüssel werden in der Regel an denselben streamingPull-Abonnentenclient gesendet. Die Affinität wird erwartet, wenn Nachrichten für einen Bestellschlüssel für einen bestimmten Abonnentenclient ausstehen. Wenn keine ausstehenden Nachrichten vorhanden sind, kann sich die Affinität aufgrund von Load Balancing oder Clienttrennungen ändern.

  Damit die Verarbeitung auch bei potenziellen Änderungen der Affinität reibungslos abläuft, ist es wichtig, dass Ihre streamingPull-Anwendung Nachrichten in jedem Client für einen bestimmten Sortierschlüssel verarbeiten kann.

• Integration in Dataflow: Aktivieren Sie die Nachrichtenreihenfolge für Abos nicht, wenn Sie Dataflow mit Pub/Sub konfigurieren. Dataflow hat einen eigenen Mechanismus für die vollständige Nachrichtenreihenfolge, der im Rahmen von Windowing-Vorgängen für die chronologische Reihenfolge aller Nachrichten sorgt. Diese Methode der Sortierung unterscheidet sich vom schlüsselbasierten Ansatz von Pub/Sub. Die Verwendung von Sortierschlüsseln mit Dataflow kann die Pipelineleistung möglicherweise verringern.

• Automatische Skalierung: Die geordnete Bereitstellung von Pub/Sub lässt sich auf Milliarden von Sortierschlüsseln skalieren. Eine größere Anzahl von Reihenfolgeschlüsseln ermöglicht eine parallele Zustellung an Abonnenten, da die Reihenfolge für alle Nachrichten mit demselben Reihenfolgeschlüssel gilt.

• Kompromisse bei der Leistung: Die geordnete Zustellung hat einige Nachteile. Im Vergleich zur ungeordneten Übermittlung verringert die geordnete Übermittlung die Verfügbarkeit von Veröffentlichungen und erhöht die End-to-End-Latenz der Nachrichtenübermittlung. Im Fall der geordneten Zustellung erfordert das Failover eine Koordination, um sicherzustellen, dass die Nachrichten in der richtigen Reihenfolge geschrieben und gelesen werden.

• Hot Key: Wenn Sie die Nachrichtenreihenfolge verwenden, werden alle Nachrichten mit demselben Reihenfolgeschlüssel in der Reihenfolge an Ihren Abonnentenclient gesendet, in der sie vom Dienst empfangen werden. Der Nutzer-Callback wird erst ausgeführt, wenn der Callback für die vorherige Nachricht abgeschlossen ist. Der maximale Durchsatz für Nachrichten mit demselben Reihenfolgeschlüssel bei der Zustellung an Abonnenten wird nicht durch Pub/Sub , sondern durch die Verarbeitungsgeschwindigkeit des Abonnentenclients begrenzt. Ein Hotkey tritt auf, wenn sich bei einem einzelnen Sortierschlüssel ein Rückstand bildet, weil die Anzahl der pro Sekunde erstellten Nachrichten die Anzahl der Nachrichten übersteigt, die der Abonnent pro Sekunde verarbeiten kann. Um Hotkeys zu vermeiden, sollten Sie die detailliertesten verfügbaren Schlüssel verwenden und die Verarbeitungszeit pro Nachricht minimieren. Sie können auch den Messwert subscription/oldest_unacked_message_age auf einen steigenden Wert hin überwachen, der auf einen Hotkey hindeuten könnte.

Weitere Informationen zur Verwendung der Nachrichtensortierung finden Sie in den folgenden Best Practices:

• Best Practices für die geordnete Nachrichtenübermittlung bei der Veröffentlichung

• Best Practices für die geordnete Nachrichtenübermittlung beim Abonnieren

Abonnentenverhalten bei der Nachrichtenreihenfolge

Abonnentenclients empfangen Nachrichten in der Reihenfolge, in der sie in einer bestimmten Region veröffentlicht wurden. Pub/Sub unterstützt verschiedene Möglichkeiten zum Empfangen von Nachrichten, z. B. Abonnentenclients, die mit Pull- und Push-Abos verbunden sind. Die Clientbibliotheken verwenden „streamingPull“ (mit Ausnahme von PHP).

Weitere Informationen zu diesen Abotypen finden Sie unter Abo auswählen.

In den folgenden Abschnitten wird erläutert, was der Empfang von Nachrichten in der richtigen Reihenfolge für die einzelnen Arten von Abonnentenclients bedeutet.

StreamingPull-Abonnentenclients

Wenn Sie die Clientbibliotheken mit „streamingPull“ verwenden, müssen Sie ein Nutzer-Callback angeben, das immer dann ausgeführt wird, wenn ein Abonnentenclient eine Nachricht empfängt. Bei Clientbibliotheken wird der Callback für einen bestimmten Reihenfolgenschlüssel für Nachrichten in der richtigen Reihenfolge vollständig ausgeführt. Wenn die Nachrichten in diesem Callback bestätigt werden, erfolgen alle Berechnungen für eine Nachricht in der richtigen Reihenfolge. Wenn im Nutzer-Callback andere asynchrone Vorgänge für Nachrichten geplant werden, muss der Abonnentenclient dafür sorgen, dass die asynchronen Vorgänge in der richtigen Reihenfolge ausgeführt werden. Eine Möglichkeit besteht darin, Nachrichten einer lokalen Arbeitswarteschlange hinzuzufügen, die in der Reihenfolge verarbeitet wird.

Pull-Abonnentenclients

Für Abonnentenclients, die mit Pull-Abos verbunden sind, unterstützt die Pub/Sub-Nachrichtenreihenfolge Folgendes:

• Alle Nachrichten für einen Bestellschlüssel in der PullResponse sind in der Liste in der richtigen Reihenfolge.

• Es kann jeweils nur ein Batch mit Nachrichten für einen Bestellschlüssel ausstehend sein.

Die Anforderung, dass jeweils nur ein Batch von Nachrichten ausstehend sein darf, ist erforderlich, um die geordnete Zustellung aufrechtzuerhalten, da der Pub/Sub-Dienst den Erfolg oder die Latenz der Antwort, die er für die Pull-Anfrage eines Abonnenten sendet, nicht garantieren kann.

Push-Abonnentenclients

Die Einschränkungen für Push sind noch strenger als für Pull. Bei einem Push-Abo unterstützt Pub/Sub jeweils nur eine ausstehende Nachricht für jeden Reihenfolgeschlüssel. Jede Nachricht wird als separate Anfrage an einen Push-Endpunkt gesendet. Das parallele Senden der Anfragen würde dasselbe Problem verursachen wie das gleichzeitige Abrufen von Abonnenten durch die Bereitstellung mehrerer Batches von Nachrichten für denselben Reihenfolgeschlüssel. Push-Abos sind möglicherweise nicht die beste Wahl für Themen, bei denen Nachrichten häufig mit demselben Sortierungsschlüssel veröffentlicht werden oder bei denen die Latenz extrem wichtig ist.

Beauftragte Käufer mit Abo exportieren

Beim Export von Abos werden geordnete Nachrichten unterstützt. Bei BigQuery-Abos werden Nachrichten mit demselben Reihenfolgeschlüssel in der Reihenfolge in ihre BigQuery-Tabelle geschrieben. Bei Cloud Storage-Abos werden Nachrichten mit demselben Reihenfolgeschlüssel möglicherweise nicht alle in dieselbe Datei geschrieben. Innerhalb derselben Datei sind Nachrichten für einen Sortierungsschlüssel in der richtigen Reihenfolge. Wenn sie auf mehrere Dateien verteilt sind, können spätere Nachrichten für einen Sortierschlüssel in einer Datei mit einem Namen angezeigt werden, der einen früheren Zeitstempel als der Zeitstempel im Namen der Datei mit den früheren Nachrichten hat.

Nachrichtenreihenfolge aktivieren

Damit die Nachrichten der Reihe nach empfangen werden, legen Sie das Attribut für die Nachrichtenreihenfolge für das Abo fest, über das Sie Nachrichten erhalten. Das Empfangen von Nachrichten der Reihe nach kann die Latenz erhöhen. Sie können die Property für die Nachrichtenreihenfolge nicht mehr ändern, nachdem Sie ein Abo erstellt haben.

Sie können das Attribut für die Nachrichtenreihenfolge festlegen, wenn Sie ein Abo in der Trusted Cloud -Konsole, mit dem Google Cloud CLI oder der Pub/Sub API erstellen.

gcloud pubsub subscriptions create SUBSCRIPTION_ID \
  --enable-message-ordering

Created subscription [SUBSCRIPTION_ID].

PUT https://pubsub.googleapis.com/v1/projects/PROJECT_ID/subscriptions/SUBSCRIPTION_ID
Authorization: Bearer $(gcloud auth application-default print-access-token)

{
  "topic": TOPIC_ID,
  "enableMessageOrdering": true,
}

{
  "name": projects/PROJECT_ID/subscriptions/SUBSCRIPTION_ID,
  "topic": projects/PROJECT_ID/topics/TOPIC_ID,
  "enableMessageOrdering": true,
}

namespace pubsub = ::google::cloud::pubsub;
namespace pubsub_admin = ::google::cloud::pubsub_admin;
[](pubsub_admin::SubscriptionAdminClient client,
   std::string const& project_id, std::string const& topic_id,
   std::string const& subscription_id) {
  google::pubsub::v1::Subscription request;
  request.set_name(
      pubsub::Subscription(project_id, subscription_id).FullName());
  request.set_topic(pubsub::Topic(project_id, topic_id).FullName());
  request.set_enable_message_ordering(true);
  auto sub = client.CreateSubscription(request);
  if (sub.status().code() == google::cloud::StatusCode::kAlreadyExists) {
    std::cout << "The subscription already exists\n";
    return;
  }
  if (!sub) throw std::move(sub).status();

  std::cout << "The subscription was successfully created: "
            << sub->DebugString() << "\n";
}

using Google.Cloud.PubSub.V1;
using Grpc.Core;

public class CreateSubscriptionWithOrderingSample
{
    public Subscription CreateSubscriptionWithOrdering(string projectId, string topicId, string subscriptionId)
    {
        SubscriberServiceApiClient subscriber = SubscriberServiceApiClient.Create();
        var topicName = TopicName.FromProjectTopic(projectId, topicId);
        var subscriptionName = SubscriptionName.FromProjectSubscription(projectId, subscriptionId);

        var subscriptionRequest = new Subscription
        {
            SubscriptionName = subscriptionName,
            TopicAsTopicName = topicName,
            EnableMessageOrdering = true
        };

        Subscription subscription = null;
        try
        {
            subscription = subscriber.CreateSubscription(subscriptionRequest);
        }
        catch (RpcException e) when (e.Status.StatusCode == StatusCode.AlreadyExists)
        {
            // Already exists.  That's fine.
        }
        return subscription;
    }
}

import (
	"context"
	"fmt"
	"io"

	"cloud.google.com/go/pubsub/v2"
	"cloud.google.com/go/pubsub/v2/apiv1/pubsubpb"
)

func createWithOrdering(w io.Writer, projectID, topic, subscription string) error {
	// projectID := "my-project-id"
	// topic := "projects/my-project-id/topics/my-topic"
	// subscription := "projects/my-project/subscriptions/my-sub"
	ctx := context.Background()
	client, err := pubsub.NewClient(ctx, projectID)
	if err != nil {
		return fmt.Errorf("pubsub.NewClient: %w", err)
	}
	defer client.Close()

	// Message ordering can only be set when creating a subscription.
	sub, err := client.SubscriptionAdminClient.CreateSubscription(ctx, &pubsubpb.Subscription{
		Name:                  subscription,
		Topic:                 topic,
		EnableMessageOrdering: true,
	})
	if err != nil {
		return fmt.Errorf("CreateSubscription: %w", err)
	}
	fmt.Fprintf(w, "Created subscription: %v\n", sub)
	return nil
}

import com.google.cloud.pubsub.v1.SubscriptionAdminClient;
import com.google.pubsub.v1.ProjectSubscriptionName;
import com.google.pubsub.v1.ProjectTopicName;
import com.google.pubsub.v1.Subscription;
import java.io.IOException;

public class CreateSubscriptionWithOrdering {
  public static void main(String... args) throws Exception {
    // TODO(developer): Replace these variables before running the sample.
    String projectId = "your-project-id";
    String topicId = "your-topic-id";
    String subscriptionId = "your-subscription-id";

    createSubscriptionWithOrderingExample(projectId, topicId, subscriptionId);
  }

  public static void createSubscriptionWithOrderingExample(
      String projectId, String topicId, String subscriptionId) throws IOException {
    try (SubscriptionAdminClient subscriptionAdminClient = SubscriptionAdminClient.create()) {

      ProjectTopicName topicName = ProjectTopicName.of(projectId, topicId);
      ProjectSubscriptionName subscriptionName =
          ProjectSubscriptionName.of(projectId, subscriptionId);

      Subscription subscription =
          subscriptionAdminClient.createSubscription(
              Subscription.newBuilder()
                  .setName(subscriptionName.toString())
                  .setTopic(topicName.toString())
                  // Set message ordering to true for ordered messages in the subscription.
                  .setEnableMessageOrdering(true)
                  .build());

      System.out.println("Created a subscription with ordering: " + subscription.getAllFields());
    }
  }
}

/**
 * TODO(developer): Uncomment these variables before running the sample.
 */
// const topicNameOrId = 'YOUR_TOPIC_NAME_OR_ID';
// const subscriptionNameOrId = 'YOUR_SUBSCRIPTION_NAME_OR_ID';

// Imports the Google Cloud client library
const {PubSub} = require('@google-cloud/pubsub');

// Creates a client; cache this for further use
const pubSubClient = new PubSub();

async function createSubscriptionWithOrdering(
  topicNameOrId,
  subscriptionNameOrId,
) {
  // Creates a new subscription
  await pubSubClient
    .topic(topicNameOrId)
    .createSubscription(subscriptionNameOrId, {
      enableMessageOrdering: true,
    });
  console.log(
    `Created subscription ${subscriptionNameOrId} with ordering enabled.`,
  );
  console.log(
    'To process messages in order, remember to add an ordering key to your messages.',
  );
}

/**
 * TODO(developer): Uncomment these variables before running the sample.
 */
// const topicNameOrId = 'YOUR_TOPIC_NAME_OR_ID';
// const subscriptionNameOrId = 'YOUR_SUBSCRIPTION_NAME_OR_ID';

// Imports the Google Cloud client library
import {PubSub} from '@google-cloud/pubsub';

// Creates a client; cache this for further use
const pubSubClient = new PubSub();

async function createSubscriptionWithOrdering(
  topicNameOrId: string,
  subscriptionNameOrId: string,
) {
  // Creates a new subscription
  await pubSubClient
    .topic(topicNameOrId)
    .createSubscription(subscriptionNameOrId, {
      enableMessageOrdering: true,
    });
  console.log(
    `Created subscription ${subscriptionNameOrId} with ordering enabled.`,
  );
  console.log(
    'To process messages in order, remember to add an ordering key to your messages.',
  );
}

from google.cloud import pubsub_v1

# TODO(developer): Choose an existing topic.
# project_id = "your-project-id"
# topic_id = "your-topic-id"
# subscription_id = "your-subscription-id"

publisher = pubsub_v1.PublisherClient()
subscriber = pubsub_v1.SubscriberClient()
topic_path = publisher.topic_path(project_id, topic_id)
subscription_path = subscriber.subscription_path(project_id, subscription_id)

with subscriber:
    subscription = subscriber.create_subscription(
        request={
            "name": subscription_path,
            "topic": topic_path,
            "enable_message_ordering": True,
        }
    )
    print(f"Created subscription with ordering: {subscription}")

# topic_id        = "your-topic-id"
# subscription_id = "your-subscription-id"

pubsub = Google::Cloud::PubSub.new
subscription_admin = pubsub.subscription_admin

subscription = subscription_admin.create_subscription \
  name: pubsub.subscription_path(subscription_id),
  topic: pubsub.topic_path(topic_id),
  enable_message_ordering: true

puts "Pull subscription #{subscription_id} created with message ordering."

Nächste Schritte

• Blogpost zur geordneten Bereitstellung lesen

• Informationen zum Fortsetzen der Veröffentlichung von Nachrichten bei nicht wiederholbaren Fehlern finden Sie unter Anfragen mit Sortierungsschlüsseln wiederholen.

• Abo überwachen

• Weitere Informationen zum Veröffentlichen von Nachrichten mit Sortierungsschlüsseln