Messaggi di ordine

L'ordinamento dei messaggi è una funzionalità di Pub/Sub che consente di ricevere i messaggi nei tuoi clienti sottoscrittori nell'ordine in cui sono stati pubblicati clienti publisher.

Ad esempio, supponiamo che un client publisher in una regione pubblichi i messaggi 1, 2, e 3 in ordine. Con l'ordinamento dei messaggi, il client sottoscrittore riceve i messaggi pubblicati nello stesso ordine. Per essere pubblicati in ordine, il client editore deve pubblicare i messaggi nella stessa regione.

L'ordinamento dei messaggi è una funzionalità utile per scenari come la registrazione delle modifiche al database, il monitoraggio delle sessioni utente e le applicazioni di streaming in cui è importante preservare la cronologia degli eventi.

Questa pagina spiega il concetto di ordinamento dei messaggi e come configurare i client sottoscrittori in modo che ricevano i messaggi in ordine. Per configurare il publisher per l'ordinamento dei messaggi, consulta Utilizzare le chiavi di ordinamento per pubblicare un .

Panoramica dell'ordinamento dei messaggi

L'ordinamento in Pub/Sub è determinato da quanto segue:

• Chiave di ordinamento: si tratta di una stringa utilizzata nei metadati dei messaggi Pub/Sub e rappresenta l'entità per la quale i messaggi devono essere ordinati. La chiave di ordinamento può avere una lunghezza massima di 1 KB. Per ricevere un insieme di messaggi ordinati in una regione, devi pubblicare tutti i messaggi con la stessa chiave di ordinamento nella stessa regione. Alcuni esempi di chiavi di ordinamento sono gli ID cliente e la chiave primaria di una riga in un database.

  La velocità effettiva di pubblicazione per ogni chiave di ordinamento è limitata a 1 MB/s. La la velocità effettiva di tutte le chiavi di ordinamento di un argomento è limitata alla quota disponibili in una regione di pubblicazione. Questo limite può essere aumentato a molte unità di GBps.

  Una chiave di ordinamento non è equivalente a una partizione in un sistema di messaggistica basato su partizioni, poiché le chiavi di ordinamento dovrebbero avere una cardinalità molto più elevata rispetto alle partizioni.

• Attiva ordinamento messaggi: si tratta di un'impostazione di abbonamento. Quando per un abbonamento è abilitato l'ordinamento dei messaggi, i client degli abbonati ricevono i messaggi pubblicati nella stessa regione con la stessa chiave di ordinamento nell'ordine in cui sono stati ricevuti dal servizio. Devi attivare questa impostazione. contenuti nell'abbonamento.

  Supponiamo che tu abbia due abbonamenti A e B associati allo stesso argomento T. La sottoscrizione A è configurata con l'ordinamento dei messaggi abilitato e la sottoscrizione B è configurata senza l'ordinamento dei messaggi abilitato. In questa architettura, sia le sottoscrizioni A che B ricevono lo stesso insieme di messaggi dall'argomento T. Se pubblichi messaggi con chiavi di ordinamento nella stessa regione, l'abbonamento A riceve i messaggi nell'ordine in cui sono stati pubblicati. Invece, la sottoscrizione B riceve i messaggi senza un ordine prestabilito.

In generale, se la tua soluzione richiede ai clienti publisher di inviare sia i prodotti ordinati che per i messaggi non ordinati, creare argomenti separati, uno per i messaggi ordinati e e gli altri per i messaggi non ordinati.

Considerazioni sull'utilizzo della messaggistica ordinata

Il seguente elenco contiene informazioni importanti sul comportamento messaggi ordinati in Pub/Sub:

• Ordinamento all'interno della chiave: i messaggi pubblicati con la stessa chiave di ordinamento devono essere ricevuti in ordine. Supponiamo che per ordinare la chiave A, pubblicare i messaggi 1, 2 e 3. Se l'ordinamento è abilitato, si prevede che 1 consegnate prima del 2, mentre la consegna 2 è prevista prima della 3.

• Ordinamento tra chiavi: i messaggi pubblicati con chiavi di ordinamento diverse non devono essere ricevuti in ordine. Supponiamo che tu abbia le chiavi di ordinamento A e B. Per ordinare la chiave A, i messaggi 1 e 2 vengono pubblicati in ordine. Per ordini chiave B, i messaggi 3 e 4 sono pubblicati in ordine. Tuttavia, il messaggio 1 potrebbe arrivi prima o dopo il messaggio 4.

• Riconsegna dei messaggi: Pub/Sub consegna ogni messaggio almeno una volta, pertanto il servizio Pub/Sub potrebbe riconsegnare i messaggi. Nuovi caricamenti di attiva la riconsegna di tutti i messaggi successivi per quella chiave, anche quelli riconosciuti. Supponiamo che un client sottoscrittore riceva i messaggi 1, 2 e 3 per una chiave di ordinamento specifica. Se il messaggio 2 viene recapitato di nuovo (perché la scadenza della conferma è scaduta o la conferma secondo il criterio del "best effort" non è stata conservata in Pub/Sub), viene recapitato di nuovo anche il messaggio 3. Se in un abbonamento sono abilitati sia l'ordinamento dei messaggi che un argomento messaggi non recapitabili, questo comportamento potrebbe non essere valido perché Pub/Sub inoltra i messaggi agli argomenti messaggi non recapitabili secondo il criterio del "best effort".

• Ritardi nell'invio di conferme e argomenti messaggi non recapitabili: i messaggi non confermati per una determinata chiave di ordinamento possono potenzialmente ritardare la consegna dei messaggi per altre chiavi di ordinamento, in particolare durante i riavvii del server o le variazioni del traffico. Per mantenere l'ordine tra questi eventi, garantire un riconoscimento tempestivo di tutti messaggi. Se non è possibile ricevere una conferma in modo tempestivo, considera l'utilizzo di un argomento messaggi non recapitabili per impedire la conservazione a tempo indeterminato dei messaggi. Tieni presente che l'ordine potrebbero non essere conservati quando i messaggi vengono scritti in un argomento messaggi non recapitabili.

• Affinità messaggio (client flussoingPull): i messaggi per la stessa chiave vengono di solito inviate allo stesso client sottoscrittore streamingPull. L'affinità è previsto quando i messaggi sono in sospeso per una chiave di ordinamento di un abbonamento. Se non ci sono messaggi in sospeso, l'affinità può per il bilanciamento del carico o la disconnessione del client.

  Per garantire un'elaborazione senza problemi anche in caso di potenziali modifiche all'affinità, fondamentale per progettare la tua applicazione flussoPull in modo che possa gestire messaggi in qualsiasi client per una determinata chiave di ordinamento.

• Integrazione con Dataflow: non abilitare l'ordinamento dei messaggi per gli abbonamenti durante la configurazione di Dataflow con in Pub/Sub. Dataflow ha un proprio meccanismo per l'ordinamento totale dei messaggi, garantendo l'ordine cronologico di tutti i messaggi come parte delle operazioni di applicazione di finestre. Questo metodo di ordinamento è diverso da Approccio basato su chiavi di ordinamento di Pub/Sub. Utilizzo delle chiavi di ordinamento e Dataflow possono ridurre le prestazioni della pipeline.

• Scalabilità automatica: la pubblicazione ordinata di Pub/Sub è scalabile fino a miliardi di chiavi di ordinamento. Un numero maggiore di chiavi di ordinamento consente consegna parallela ai sottoscrittori poiché l'ordinamento si applica a tutti i messaggi con la stessa chiave di ordinamento.

La consegna dell'ordine comporta alcuni compromessi. Rispetto a non ordinato pubblicazione, la consegna ordinata potrebbe ridurre leggermente la disponibilità di pubblicazione la latenza di recapito end-to-end dei messaggi. Nel caso di consegna ordinato, il failover richiede il coordinamento per garantire che i messaggi vengano scritti e letti nell'ordine corretto.

Per ulteriori informazioni su come utilizzare l'ordinamento dei messaggi, vedi quanto segue argomenti delle best practice:

• Best practice per la messaggistica ordinata nella pubblicazione

• Best practice per la messaggistica ordinata nelle iscrizioni

Comportamento del client sottoscrittore per l'ordinamento dei messaggi

I client sottoscrittori ricevono i messaggi nell'ordine in cui sono stati pubblicati in una regione specifica. Pub/Sub supporta diversi modi per ricevere messaggi, ad esempio client sottoscrittori collegati a sottoscrizioni pull e push. Le librerie client utilizzano streamingPull (ad eccezione di PHP).

Per scoprire di più su questi tipi di abbonamento, consulta l'articolo Scegliere un tipo di abbonamento.

Le sezioni seguenti illustrano cosa significa ricevere i messaggi in ordine per ciascun tipo di client di abbonato.

Client sottoscrittori StreamingPull

Quando utilizzi le librerie client con streamingPull, devi specificare un utente che viene eseguito ogni volta che un messaggio viene ricevuto da un client sottoscrittore. Con le librerie client, per una determinata chiave di ordinamento, il callback viene eseguito fino al completamento dei messaggi nell'ordine corretto. Se i messaggi vengono riconosciuti all'interno del callback, vengono eseguiti tutti i calcoli in ordine. Tuttavia, se il callback dell'utente pianifica altro lavoro asincrono sui messaggi, il client dell'abbonato deve assicurarsi che il lavoro asincrono venga eseguito in ordine. Un'opzione è aggiungere messaggi a una coda di lavoro locale che viene elaborata in ordine.

Client sottoscrittori di tipo pull

Per i client sottoscrittori connessi alle sottoscrizioni pull, Pub/Sub l'ordinamento dei messaggi supporta quanto segue:

• Tutti i messaggi per una chiave di ordinamento in PullResponse sono nell'ordine corretto nell'elenco.

• Solo un batch di messaggi può essere in sospeso per una chiave di ordinamento alla volta.

Il requisito che solo un batch di messaggi possa essere in sospeso alla volta è necessario per mantenere la consegna ordinata, poiché il servizio Pub/Sub non può garantire il successo o la latenza della risposta inviata per la pull request di un sottoscrittore.

Client sottoscrittori push

I limiti alla spinta sono ancora più rigidi di quelli alla spinta. Per una sottoscrizione push, Pub/Sub supporta un solo elemento in sospeso per ogni chiave di ordinamento alla volta. Ogni messaggio viene inviato a un endpoint push come richiesta separata. Pertanto, l'invio delle richieste in parallelo presenterebbe lo stesso problema dell'invio di più batch di messaggi per la stessa chiave di ordinamento per estrarre contemporaneamente gli iscritti. Le sottoscrizioni push potrebbero non essere una buona scelta per gli argomenti in cui i messaggi vengono vengono spesso pubblicati con la stessa chiave di ordinamento o laddove la latenza estremamente importante.

Esportare i client sottoscrittori

Le iscrizioni per l'esportazione supportano i messaggi in ordine. Per BigQuery sottoscrizioni, i messaggi con la stessa chiave di ordinamento vengono scritti nel relativo BigQuery in ordine. Per gli abbonamenti a Cloud Storage, i messaggi con la stessa chiave di ordinamento potrebbero non essere scritti tutti nello stesso file. All'interno dello stesso file, i messaggi relativi a una chiave di ordinamento sono in ordine. Quando distribuiti in più file, i messaggi successivi per una chiave di ordinamento possono essere visualizzati in con un nome la cui data e ora è precedente al timestamp nel nome il file con i messaggi precedenti.

Attivare l'ordinamento dei messaggi

Per ricevere i messaggi in ordine, imposta la proprietà di ordinamento nella sezione abbonamento da cui ricevi i messaggi. La ricezione dei messaggi in ordine potrebbe aumentare la latenza. Non puoi modificare la proprietà di ordinamento dei messaggi dopo aver creato un abbonamento.

Puoi impostare la proprietà di ordinamento nella sottoscrizione al momento di creare una sottoscrizione utilizzando la console Google Cloud, Google Cloud CLI o l'API Pub/Sub.

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."

Passaggi successivi

• Leggi il post del blog sulla pubblicazione ordinata.

• Per continuare a pubblicare i messaggi in caso di errori non ripetibili, consulta Rieseguire le richieste con chiavi di ordinamento.

• Monitora abbonamento.

• Scopri di più sulla pubblicazione di messaggi con chiavi di ordinamento.