Messages de commande

L'ordre des messages est une fonctionnalité de Pub/Sub qui vous permet de recevoir les messages dans vos clients abonnés dans l'ordre dans lequel ils ont été publiés par les clients éditeurs.

Par exemple, supposons qu'un client éditeur d'une région publie les messages 1, 2 et 3 dans l'ordre. Avec le tri des messages, le client abonné reçoit les messages publiés dans le même ordre. Pour que les messages soient distribués dans l'ordre, le client éditeur doit les publier dans la même région. Toutefois, les abonnés peuvent se connecter à n'importe quelle région et la garantie de commande est toujours maintenue.

L'ordre des messages est une fonctionnalité utile pour les scénarios tels que la capture des modifications de base de données, le suivi des sessions utilisateur et les applications de streaming où il est important de préserver la chronologie des événements.

Cette page explique le concept d'ordre des messages et comment configurer vos clients abonnés pour recevoir les messages dans l'ordre. Pour configurer vos clients éditeurs pour le tri des messages, consultez Utiliser des clés de tri pour publier un message.

Présentation du tri des messages

L'ordre des messages dans Pub/Sub est déterminé par les éléments suivants :

• Clé de tri : chaîne utilisée dans les métadonnées des messages Pub/Sub et représentant l'entité pour laquelle les messages doivent être triés. La clé de tri peut comporter jusqu'à 1 Ko. Pour recevoir un ensemble de messages ordonnés dans une région, vous devez publier tous les messages avec la même clé de tri dans la même région. Voici quelques exemples de clés de tri : les ID client et la clé primaire d'une ligne dans une base de données.

  Le débit de publication sur chaque clé de tri est limité à 1 Mo/s. Le débit pour toutes les clés de commande d'un sujet est limité au quota disponible dans une région de publication. Cette limite peut être augmentée à de nombreuses unités de GBps.

  Une clé de tri n'est pas équivalente à une partition dans un système de messagerie basé sur des partitions, car les clés de tri sont censées avoir une cardinalité beaucoup plus élevée que les partitions.

• Activer le tri des messages : il s'agit d'un paramètre d'abonnement. Lorsqu'un abonnement est associé au tri des messages, les clients abonnés reçoivent les messages publiés dans la même région avec la même clé de tri dans l'ordre dans lequel ils ont été reçus par le service. Vous devez activer ce paramètre dans l'abonnement.

  Imaginons que vous ayez deux abonnements A et B associés au même sujet T. L'abonnement A est configuré avec le tri des messages activé, tandis que l'abonnement B est configuré sans le tri des messages activé. Dans cette architecture, les abonnements A et B reçoivent le même ensemble de messages du sujet T. Si vous publiez des messages avec des clés de tri dans la même région, l'abonnement A reçoit les messages dans l'ordre dans lequel ils ont été publiés. L'abonnement B reçoit les messages sans ordre particulier.

En général, si votre solution nécessite que les clients éditeurs envoient des messages triés et non triés, créez des sujets distincts : un pour les messages triés et un autre pour les messages non triés.

Éléments à prendre en compte lors de l'utilisation de la messagerie ordonnée

La liste suivante contient des informations importantes concernant le comportement de la messagerie ordonnée dans Pub/Sub :

• Tri dans une même clé : les messages publiés avec la même clé de tri doivent être reçus dans l'ordre. Supposons que vous publiez les messages 1, 2 et 3 pour la clé de tri A. Si le tri est activé, 1 doit être fourni avant 2 et 2 avant 3.

• Tri entre les clés : les messages publiés avec des clés de tri différentes ne sont pas censés être reçus dans l'ordre. Supposons que vous disposiez des clés de commande A et B. Pour la clé de tri A, les messages 1 et 2 sont publiés dans l'ordre. Pour la clé de tri B, les messages 3 et 4 sont publiés dans l'ordre. Toutefois, le message 1 peut arriver avant ou après le message 4.

• Nouvelle distribution des messages : Pub/Sub distribue chaque message au moins une fois. Le service Pub/Sub peut donc distribuer plusieurs fois des messages. La redistribution d'un message déclenche la redistribution de tous les messages ultérieurs pour cette clé, même ceux qui ont été confirmés. Supposons qu'un client abonné reçoive les messages 1, 2 et 3 pour une clé de tri spécifique. Si le message 2 est redistribué (parce que le délai de confirmation a expiré ou que la confirmation de la meilleure façon possible n'a pas été conservée dans Pub/Sub), le message 3 l'est également. Si le tri des messages et une file d'attente de lettres mortes sont activés sur un abonnement, ce comportement n'est pas garanti, car Pub/Sub transfère les messages vers des files d'attente de lettres mortes de la manière la plus optimale possible.

• Retards d'accusé de réception et sujets de lettres mortes : les messages non confirmés pour une clé de tri donnée peuvent retarder la distribution des messages pour d'autres clés de tri, en particulier lors des redémarrages de serveur ou des changements de trafic. Pour maintenir l'ordre lors de tels événements, assurez-vous de confirmer la réception de tous les messages en temps voulu. Si une confirmation rapide n'est pas possible, envisagez d'utiliser un file d'attente de lettres mortes pour éviter la conservation indéfinie des messages. Notez que l'ordre peut ne pas être conservé lorsque des messages sont écrits dans un file d'attente de lettres mortes.

• Affinité des messages (clients streamingPull) : les messages associés à une même clé sont généralement distribués au même client abonné streamingPull. L'affinité est attendue lorsque des messages sont en attente pour une clé de tri pour un client abonné spécifique. S'il n'y a pas de messages en attente, l'affinité peut changer pour l'équilibrage de charge ou les déconnexions des clients.

  Pour garantir un traitement fluide même en cas de changement d'affinité potentiel, il est essentiel de concevoir votre application streamingPull de manière à ce qu'elle puisse gérer les messages dans n'importe quel client pour une clé de tri donnée.

• Intégration à Dataflow : n'activez pas le tri des messages pour les abonnements lorsque vous configurez Dataflow avec Pub/Sub. Dataflow possède son propre mécanisme de tri des messages, qui garantit l'ordre chronologique de tous les messages dans les opérations de fenêtrage. Cette méthode de tri diffère de l'approche basée sur les clés de tri de Pub/Sub. L'utilisation de clés de tri avec Dataflow peut potentiellement réduire les performances du pipeline.

• Scaling automatique : la diffusion ordonnée de Pub/Sub s'adapte à des milliards de clés de tri. Un plus grand nombre de clés de tri permet une diffusion plus parallèle aux abonnés, car le tri s'applique à tous les messages ayant la même clé de tri.

• Compromis en termes de performances : la livraison ordonnée présente certains inconvénients. Par rapport à la distribution non ordonnée, la distribution ordonnée réduit la disponibilité de la publication et augmente la latence de distribution des messages de bout en bout. Dans le cas d'une distribution ordonnée, le basculement nécessite une coordination pour s'assurer que les messages sont écrits et lus dans le bon ordre.

• Touche d'accès rapide : lorsque vous utilisez le tri des messages, tous les messages ayant la même clé de tri sont envoyés à votre client abonné dans l'ordre dans lequel ils sont reçus par le service. Le rappel utilisateur ne s'exécute pas tant que le rappel n'est pas terminé pour le message précédent. Le débit maximal des messages partageant la même clé de tri lors de la distribution aux abonnés n'est pas limité par Pub/Sub , mais par la vitesse de traitement de votre client abonné. Une clé de commande "hot" se produit lorsqu'un backlog s'accumule sur une clé de commande individuelle, car le nombre de messages produits par seconde dépasse le nombre de messages que l'abonné peut traiter par seconde. Pour atténuer les clés chaudes, utilisez les clés les plus granulaires possibles et minimisez le temps de traitement par message. Vous pouvez également surveiller la métrique subscription/oldest_unacked_message_age pour détecter une valeur en hausse, ce qui peut indiquer une clé active.

Pour en savoir plus sur l'utilisation de l'ordre des messages, consultez les articles suivants sur les bonnes pratiques :

• Bonnes pratiques pour les messages ordonnés dans la publication

• Bonnes pratiques pour la messagerie ordonnée lors de l'abonnement

Comportement du client abonné pour l'ordre des messages

Les clients abonnés reçoivent les messages dans l'ordre dans lequel ils ont été publiés dans une région spécifique. Pub/Sub est compatible avec différentes méthodes de réception des messages, comme les clients abonnés connectés aux abonnements pull et push. Les bibliothèques clientes utilisent streamingPull (à l'exception de PHP).

Pour en savoir plus sur ces types d'abonnement, consultez Choisir un type d'abonnement.

Les sections suivantes expliquent ce que signifie la réception des messages dans l'ordre pour chaque type de client abonné.

Clients abonnés StreamingPull

Lorsque vous utilisez les bibliothèques clientes avec streamingPull, vous devez spécifier un rappel utilisateur qui s'exécute chaque fois qu'un message est reçu par un client abonné. Avec les bibliothèques clientes, pour une clé de tri donnée, le rappel est exécuté jusqu'à la fin sur les messages dans le bon ordre. Si les messages sont reconnus dans ce rappel, tous les calculs sur un message se produisent dans l'ordre. Toutefois, si le rappel utilisateur planifie d'autres tâches asynchrones sur les messages, le client abonné doit s'assurer que les tâches asynchrones sont effectuées dans l'ordre. Une option consiste à ajouter des messages à une file d'attente de tâches locale qui est traitée dans l'ordre.

Clients abonnés pull

Pour les clients abonnés connectés à des abonnements pull, le tri des messages Pub/Sub est compatible avec les éléments suivants :

• Tous les messages associés à une clé de tri dans PullResponse sont dans le bon ordre dans la liste.

• Un seul lot de messages peut être en attente pour une clé de commande à la fois.

L'exigence selon laquelle une seule série de messages peut être en attente à la fois est nécessaire pour maintenir la distribution ordonnée, car le service Pub/Sub ne peut pas garantir le succès ni la latence de la réponse qu'il envoie pour la demande d'extraction d'un abonné.

Clients abonnés push

Les restrictions concernant le push sont encore plus strictes que celles concernant le pull. Pour un abonnement push, Pub/Sub n'accepte qu'un seul message en attente pour chaque clé de tri à la fois. Chaque message est envoyé à un point de terminaison push sous la forme d'une requête distincte. Par conséquent, l'envoi des requêtes en parallèle poserait le même problème que l'envoi de plusieurs lots de messages pour la même clé de tri aux abonnés simultanément. Les abonnements push ne sont peut-être pas un bon choix pour les sujets dans lesquels les messages sont fréquemment publiés avec la même clé de tri ou lorsque la latence est extrêmement importante.

Exporter les clients abonnés

Les abonnements aux exportations sont compatibles avec les messages ordonnés. Pour les abonnements BigQuery, les messages ayant la même clé de tri sont écrits dans leur table BigQuery dans l'ordre. Pour les abonnements Cloud Storage, il est possible que les messages ayant la même clé de tri ne soient pas tous écrits dans le même fichier. Lorsque les messages d'une clé de tri se trouvent dans le même fichier, ils sont dans l'ordre. Lorsqu'ils sont répartis sur plusieurs fichiers, les messages ultérieurs pour une clé de tri peuvent apparaître dans un fichier dont le nom comporte un code temporel antérieur à celui du nom du fichier contenant les messages antérieurs.

Activer le tri des messages

Pour recevoir les messages dans l'ordre, définissez la propriété de tri des messages sur l'abonnement à partir duquel vous recevez les messages. La réception des messages dans l'ordre peut augmenter la latence. Vous ne pouvez pas modifier la propriété d'ordre des messages après avoir créé un abonnement.

Vous pouvez définir la propriété de tri des messages lorsque vous créez un abonnement à l'aide de la console Trusted Cloud , de la Google Cloud CLI ou de 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"

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

Étapes suivantes

• Consultez l'article de blog sur la diffusion ordonnée.

• Pour continuer à publier des messages en cas d'erreurs ne permettant aucune autre tentative, consultez Réessayer les requêtes avec des clés de tri.

• Surveillez votre abonnement.

• En savoir plus sur la publication de messages avec des clés de tri