Nachrichten in eine Reihenfolge bringen

Die Nachrichtenreihenfolge ist ein Pub/Sub-Feature, mit dem du Nachrichten in deinen Abonnentenclients in der Reihenfolge empfangen kannst, 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 der richtigen 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 die Nachrichten in derselben Region veröffentlichen. Abonnenten können jedoch eine Verbindung zu jeder Region herstellen und die Bestellgarantie bleibt bestehen.

Die Nachrichtensortierung ist eine nützliche Funktion für Szenarien wie die Erfassung von Datenbankänderungen, das Tracking von Nutzersitzungen und Streaminganwendungen, bei denen die chronologische Reihenfolge der Ereignisse wichtig ist.

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

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 Nachrichten in einer Region erhalten 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 aller Sortierschlüssel für ein Thema ist auf das Kontingent beschränkt, das in einer Veröffentlichungsregion verfügbar ist. Dieses Limit kann auf viele GBps erhöht werden.

  Ein Sortierschlüssel entspricht nicht einer Partition in einem partitionbasierten Messaging-System, da Sortierschlüssel eine viel höhere Kardinalität als Partitionen haben sollen.

• Nachrichtenreihenfolge aktivieren: Dies ist eine Aboeinstellung. Wenn für ein Abo die Nachrichtenreihenfolge aktiviert ist, erhalten 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. Bei Abo A ist die Nachrichtenreihenfolge aktiviert und bei Abo B nicht. In dieser Architektur erhalten sowohl Abo A als auch Abo B dieselben Nachrichten vom Thema T. Wenn Sie Nachrichten mit Reihenfolgeschlüsseln in derselben Region veröffentlichen, empfängt Abo A die Nachrichten in der Reihenfolge, in der sie veröffentlicht wurden. Abo B hingegen empfängt die Nachrichten ohne eine bestimmte Reihenfolge.

Wenn Publisher-Clients in Ihrer Lösung sowohl sortierte als auch unsortierte Nachrichten senden müssen, sollten Sie im Allgemeinen separate Themen erstellen, eines für sortierte und eines für unsortierte Nachrichten.

Überlegungen zur Verwendung von bestellbaren 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 Reihenfolgeschlüssel veröffentlicht werden, werden in der Regel in der richtigen Reihenfolge empfangen. Angenommen, Sie veröffentlichen für den Sortierungsschlüssel A die Nachrichten 1, 2 und 3. Wenn die Sortierung aktiviert ist, wird „1“ voraussichtlich vor „2“ und „2“ vor „3“ ausgeliefert.

• Sortierung über Schlüssel hinweg: Nachrichten, die mit verschiedenen Reihenfolgeschlüsseln veröffentlicht werden, werden nicht in der richtigen Reihenfolge empfangen. Angenommen, Sie haben die Bestellschlüssel A und B. Bei Sortierschlüssel A werden Nachrichten 1 und 2 nacheinander veröffentlicht. Bei Reihenfolgeschlüssel B werden die Nachrichten 3 und 4 in der richtigen Reihenfolge veröffentlicht. Nachricht 1 kann jedoch vor oder nach Nachricht 4 eintreffen.

• Wiedervorlage von Nachrichten: Pub/Sub sendet jede Nachricht mindestens einmal, sodass der Pub/Sub-Dienst Nachrichten möglicherweise wiederholt sendet. 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 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 nicht in Pub/Sub gespeichert 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: Unbestä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 Verkehrsänderungen. Um für Ordnung bei solchen Ereignissen zu sorgen, müssen Sie alle Nachrichten zeitnah bestätigen. Wenn eine zeitnahe Bestätigung nicht möglich ist, sollten Sie ein Thema für unzustellbare Nachrichten verwenden, um eine unbegrenzte Nachrichtenspeicherung zu vermeiden. Die Reihenfolge wird möglicherweise nicht beibehalten, 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. Eine Affinität wird erwartet, wenn Nachrichten für einen Reihenfolgeschlüssel an einen bestimmten Abonnentenclient ausstehend sind. Wenn keine ausstehenden Nachrichten vorhanden sind, kann die Affinität für das Load Balancing oder die Clientverbindungen geändert werden.

  Damit die Verarbeitung auch bei potenziellen Affinitätsänderungen reibungslos abläuft, muss deine StreamingPull-Anwendung so konzipiert sein, dass sie Nachrichten in jedem Client für einen bestimmten Sortierschlüssel verarbeiten kann.

• Integration mit Dataflow: Aktivieren Sie die Nachrichtensortierung für Abos nicht, wenn Sie Dataflow mit Pub/Sub konfigurieren. Dataflow hat einen eigenen Mechanismus für die Gesamtsortierung von Nachrichten, der im Rahmen von Fenstervorgängen für eine chronologische Reihenfolge aller Nachrichten sorgt. Diese Sortierungsmethode unterscheidet sich vom Pub/Sub-basierten Schlüsselansatz. Die Verwendung von Sortierschlüsseln mit Dataflow kann die Pipelineleistung möglicherweise beeinträchtigen.

• Automatische Skalierung: Die geordnete Auslieferung von Pub/Sub kann auf Milliarden von Sortierschlüsseln skaliert werden. Eine größere Anzahl von Reihenfolgeschlüsseln ermöglicht eine parallelere Zustellung an Abonnenten, da die Sortierung auf alle Nachrichten mit demselben Reihenfolgeschlüssel angewendet wird.

Die Lieferung auf Bestellung hat jedoch einige Nachteile. Im Vergleich zur unverschlüsselten Zustellung verringert die sortierte Zustellung die Veröffentlichungsverfügbarkeit und erhöht die End-to-End-Latenz der Nachrichtenübermittlung. Bei der geordneten Zustellung ist eine Abstimmung erforderlich, damit die Nachrichten in der richtigen Reihenfolge geschrieben und gelesen werden.

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

• Best Practices für angeordnete Nachrichten in der Verlagsbranche

• Best Practices für die geordnete Übermittlung von Nachrichten bei der Anmeldung

Verhalten von Abonnenten-Clients bei der Nachrichtensortierung

Abonnentenclients erhalten 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 Abotyp 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 einen Nutzer-Callback angeben, der ausgeführt wird, wenn ein Abonnentenclient eine Nachricht empfängt. Bei Clientbibliotheken wird der Callback für jeden beliebigen Reihenfolgenschlüssel bis zum Ende für Nachrichten in der richtigen Reihenfolge ausgeführt. Wenn die Nachrichten innerhalb dieses Rückrufs bestätigt werden, werden alle Berechnungen für eine Nachricht in der richtigen Reihenfolge ausgeführt. Wenn der Nutzer-Callback jedoch andere asynchrone Aufgaben für Nachrichten plant, muss der Abonnentenclient dafür sorgen, dass die asynchronen Aufgaben in der richtigen Reihenfolge ausgeführt werden. Eine Möglichkeit besteht darin, Nachrichten einer lokalen Arbeitswarteschlange hinzuzufügen, die nacheinander verarbeitet wird.

Pull-Abonnenten

Für Abonnentenclients, die mit Pull-Abos verbunden sind, wird bei der Pub/Sub-Nachrichtensortierung Folgendes unterstützt:

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

• Für einen Bestellschlüssel kann jeweils nur ein Nachrichtenbatch ausstehen.

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

Push-Abonnenten-Clients

Die Einschränkungen für Push-Nachrichten sind noch strenger als für Pull-Nachrichten. 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 Senden der Anfragen parallel würde also dasselbe Problem verursachen wie das Senden mehrerer Nachrichtenbatches für denselben Reihenfolgenschlüssel, um Abonnenten gleichzeitig abzurufen. 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.

Abonnenten-Clients exportieren

Exportierte Abos unterstützen sortierte Nachrichten. Bei BigQuery-Abos werden Nachrichten mit demselben Reihenfolgeschlüssel nacheinander in die 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 sortiert. Wenn Nachrichten auf mehrere Dateien verteilt sind, können spätere Nachrichten für einen Sortierschlüssel in einer Datei mit einem Namen erscheinen, 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 Nachrichtensortierungseigenschaft 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 Google Cloud Console, der 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"
	"time"

	"cloud.google.com/go/pubsub"
)

func createWithOrdering(w io.Writer, projectID, subID string, topic *pubsub.Topic) error {
	// projectID := "my-project-id"
	// subID := "my-sub"
	// topic of type https://godoc.org/cloud.google.com/go/pubsub#Topic
	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.CreateSubscription(ctx, subID, pubsub.SubscriptionConfig{
		Topic:                 topic,
		AckDeadline:           20 * time.Second,
		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

topic        = pubsub.topic topic_id
subscription = topic.subscribe subscription_id,
                               message_ordering: true

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

Nächste Schritte

• Blogpost zur geordneten Zustellung lesen

• Wie Sie bei nicht wiederholbaren Fehlern weiterhin Nachrichten veröffentlichen, erfahren Sie unter Anfragen mit Sortierungsschlüsseln wiederholen.

• Abo überwachen

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