Diffusion de type "exactement une fois"

Cette page explique comment recevoir et accuser réception des messages à l'aide de la fonctionnalité "exactement une fois" de Pub/Sub, qui vous permet de suivre et d'empêcher le traitement en double des messages. Lorsque la fonctionnalité est activée, Pub/Sub fournit la sémantique suivante :

• Les abonnés peuvent déterminer si les accusés de réception des messages ont été envoyés.

• Aucune nouvelle distribution n'a lieu une fois le message confirmé.

• Aucune nouvelle tentative de distribution n'est effectuée tant qu'un message est en attente. Un message est considéré comme en attente jusqu'à l'expiration du délai d'accusé de réception ou jusqu'à ce qu'il soit confirmé.

• En cas de plusieurs distributions valides, en raison de l'expiration du délai de confirmation ou d'un accusé de réception négatif initié par le client, seul le dernier ID d'accusé de réception peut être utilisé pour confirmer la réception du message. Toutes les demandes avec un ID de confirmation précédent échouent.

Lorsque l'option "exactement une fois" est activée, les abonnés peuvent s'assurer que les messages sont traités une seule fois en suivant ces consignes :

• Confirmer les messages dans le délai de confirmation.

• Conservez les informations sur l'état du traitement d'un message jusqu'à ce qu'il soit correctement confirmé.

• Utilisez les informations sur la progression du traitement d'un message pour éviter les doublons en cas d'échec de l'accusé de réception.

Seul le type d'abonnement pull est compatible avec la distribution "exactement une fois", y compris les abonnés qui utilisent l'API StreamingPull. La distribution push et les abonnements aux exportations ne sont pas compatibles avec la distribution de type "exactement une fois".

Pub/Sub est compatible avec la distribution de type "exactement une fois" dans une région cloud, basée sur un ID de message unique défini par Pub/Sub.

Versions recommandées des bibliothèques clientes

• Pour des performances optimales, utilisez la dernière version de la bibliothèque cliente : Python v2.13.6 ou version ultérieure, Java v1.139.0 ou version ultérieure, PHP v1.39.0 ou version ultérieure, C# v3.2.0 ou version ultérieure, C++ v2.1.0, Go v1.25.1 ou version ultérieure, Node v3.2.0 ou version ultérieure et Ruby v2.12.1 ou version ultérieure.

Nouvelle livraison ou doublon

Il est important de comprendre la différence entre les nouvelles distributions attendues et inattendues.

• Une redistribution peut se produire en raison d'un accusé de réception négatif d'un message initié par le client ou lorsque le client ne prolonge pas le délai d'accusé de réception du message avant son expiration. Les nouvelles livraisons sont considérées comme valides et le système fonctionne comme prévu.

  Pour résoudre les problèmes de redélivrance, consultez Gérer les doublons.

• Un message est considéré comme un doublon lorsqu'il est renvoyé après un accusé de réception ou avant l'expiration du délai d'accusé de réception.

• Un message rediffusé conserve le même ID de message entre les tentatives de rediffusion.

Les abonnements pour lesquels la distribution de type "exactement une fois" est activée ne reçoivent pas de distributions en double.

Compatibilité avec la diffusion de type "exactement une fois" dans les bibliothèques clientes

• Les bibliothèques clientes compatibles disposent d'une interface pour l'accusé de réception avec réponse (exemple : Go). Vous pouvez utiliser cette interface pour vérifier si la demande d'accusé de réception a abouti. Si la demande d'accusé de réception aboutit, les clients sont assurés de ne pas recevoir de nouvelle livraison. Si la demande d'accusé de réception échoue, les clients peuvent s'attendre à une nouvelle livraison.

• Les clients peuvent également utiliser les bibliothèques clientes compatibles sans l'interface d'accusé de réception. Toutefois, dans ce cas, les échecs d'accusé de réception peuvent entraîner une nouvelle distribution silencieuse des messages.

• Les bibliothèques clientes compatibles disposent d'interfaces permettant de définir la durée minimale de prolongation du bail (exemple : Go). Vous devez définir une valeur élevée pour la durée minimale de prolongation du bail afin d'éviter toute expiration d'accusé de réception lié au réseau. La valeur maximale est définie sur 600 secondes.

• Si vous utilisez la bibliothèque cliente Java et que vous initialisez votre abonné avec un canal gRPC personnalisé à l'aide de la méthode setChannelProvider(), nous vous recommandons également de définir maxInboundMetadataSize sur au moins 1 Mo lors de la création de votre TransportChannelProvider. Pour cette configuration, vous pouvez utiliser la méthode InstantiatingGrpcChannelProvider.Builder.setMaxInboundMetadataSize() ou ManagedChannelBuilder.maxInboundMetadataSize().

Les valeurs et la plage par défaut des variables liées à la diffusion "une seule fois", ainsi que les noms des variables, peuvent varier d'une bibliothèque cliente à l'autre. Par exemple, dans la bibliothèque cliente Java, les variables suivantes contrôlent la diffusion exacte.

Dans le cas d'une distribution de type "exactement une fois", la requête modifyAckDeadline ou acknowledgment envoyée à Pub/Sub échoue lorsque l'ID d'accusé de réception a déjà expiré. Dans ce cas, le service considère que l'ID d'accusé de réception expiré n'est pas valide, car une diffusion plus récente est peut-être déjà en cours. Cela est volontaire pour la diffusion "exactement une fois". Les requêtes acknowledgment et ModifyAckDeadline renvoient une réponse INVALID_ARGUMENT. Lorsque la diffusion de type "exactement une fois" est désactivée, ces requêtes renvoient OK en cas d'ID d'accusé de réception expirés.

Pour vous assurer que les requêtes acknowledgment et ModifyAckDeadline disposent d'ID d'accusé de réception valides, envisagez de définir la valeur de minDurationPerAckExtension sur un nombre élevé.

Points à prendre en compte concernant les régions

La garantie de distribution de type "exactement une fois" ne s'applique que lorsque les abonnés se connectent au service dans la même région. Si votre application d'abonné est répartie sur plusieurs régions, cela peut entraîner la diffusion de messages en double, même lorsque la diffusion de type "exactement une fois" est activée. Les éditeurs peuvent envoyer des messages à n'importe quelle région, et la garantie d'envoi unique est toujours respectée.

Lorsque vous exécutez votre application dans Trusted Cloud, elle se connecte par défaut au point de terminaison Pub/Sub de la même région. Par conséquent, l'exécution de votre application dans une seule région au sein de Trusted Cloudgarantit généralement que vous interagissez avec une seule région.

Lorsque vous exécutez votre application d'abonné en dehors de Trusted Cloudou dans plusieurs régions, vous pouvez vous assurer de vous connecter à une seule région en utilisant un point de terminaison géographique lorsque vous configurez votre client Pub/Sub. Tous les points de terminaison d'emplacement pour Pub/Sub pointent vers des régions uniques. Pour en savoir plus sur les points de terminaison géographiques, consultez Points de terminaison Pub/Sub. Pour obtenir la liste de tous les points de terminaison régionaux pour Pub/Sub, consultez Liste des points de terminaison régionaux.

Créer des abonnements avec la distribution de type "exactement une fois"

Vous pouvez créer un abonnement avec une diffusion exacte à l'aide de la console Trusted Cloud , de Google Cloud CLI, d'une bibliothèque cliente ou de l'API Pub/Sub.

Abonnement pull

gcloud pubsub subscriptions create SUBSCRIPTION_ID \
  --topic=TOPIC_ID \
  --enable-exactly-once-delivery

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

{
  "topic": "projects/PROJECT_ID/topics/TOPIC_ID",
  "enableExactlyOnceDelivery": 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_exactly_once_delivery(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 CreateSubscriptionWithExactlyOnceDeliverySample
{
    public Subscription CreateSubscriptionWithExactlyOnceDelivery(string projectId, string topicId, string subscriptionId)
    {
        SubscriberServiceApiClient subscriber = SubscriberServiceApiClient.Create();
        TopicName topicName = TopicName.FromProjectTopic(projectId, topicId);
        SubscriptionName subscriptionName = SubscriptionName.FromProjectSubscription(projectId, subscriptionId);

        var subscriptionRequest = new Subscription
        {
            SubscriptionName = subscriptionName,
            TopicAsTopicName = topicName,
            EnableExactlyOnceDelivery = 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 createSubscriptionWithExactlyOnceDelivery(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()

	pbSub := &pubsubpb.Subscription{
		Name:                      subscription,
		Topic:                     topic,
		EnableExactlyOnceDelivery: true,
	}
	sub, err := client.SubscriptionAdminClient.CreateSubscription(ctx, pbSub)
	if err != nil {
		return fmt.Errorf("failed to create exactly once sub: %w", err)
	}
	fmt.Fprintf(w, "Created a subscription with exactly once delivery enabled: %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 CreateSubscriptionWithExactlyOnceDelivery {
  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";

    createSubscriptionWithExactlyOnceDeliveryExample(projectId, topicId, subscriptionId);
  }

  public static void createSubscriptionWithExactlyOnceDeliveryExample(
      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())
                  // Enable exactly once delivery in the subscription.
                  .setEnableExactlyOnceDelivery(true)
                  .build());

      System.out.println(
          "Created a subscription with exactly once delivery enabled: "
              + subscription.getAllFields());
    }
  }
}

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_exactly_once_delivery": True,
        }
    )
    print(
        f"Created subscription with exactly once delivery enabled: {subscription}"
    )

/**
 * 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 createSubscriptionWithExactlyOnceDelivery(
  topicNameOrId,
  subscriptionNameOrId,
) {
  // Creates a new subscription
  await pubSubClient
    .topic(topicNameOrId)
    .createSubscription(subscriptionNameOrId, {
      enableExactlyOnceDelivery: true,
    });
  console.log(
    `Created subscription ${subscriptionNameOrId} with exactly-once delivery.`,
  );
  console.log(
    'To process messages, remember to check the return value of ackWithResponse().',
  );
}

/**
 * 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 createSubscriptionWithExactlyOnceDelivery(
  topicNameOrId: string,
  subscriptionNameOrId: string,
) {
  // Creates a new subscription
  await pubSubClient
    .topic(topicNameOrId)
    .createSubscription(subscriptionNameOrId, {
      enableExactlyOnceDelivery: true,
    });
  console.log(
    `Created subscription ${subscriptionNameOrId} with exactly-once delivery.`,
  );
  console.log(
    'To process messages, remember to check the return value of ackWithResponse().',
  );
}

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

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

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

puts "Created subscription with exactly once delivery enabled: " \
     "#{subscription_id}"

use Google\Cloud\PubSub\PubSubClient;

/**
 * Creates a Pub/Sub subscription with `Exactly Once Delivery` enabled.
 *
 * @param string $projectId  The Google project ID.
 * @param string $topicName  The Pub/Sub topic name.
 * @param string $subscriptionName  The Pub/Sub subscription name.
 */
function create_subscription_with_exactly_once_delivery(
    string $projectId,
    string $topicName,
    string $subscriptionName
): void {
    $pubsub = new PubSubClient([
        'projectId' => $projectId,
    ]);
    $topic = $pubsub->topic($topicName);
    $subscription = $topic->subscription($subscriptionName);
    $subscription->create([
        'enableExactlyOnceDelivery' => true
    ]);

    // Exactly Once Delivery status for the subscription
    $status = $subscription->info()['enableExactlyOnceDelivery'];

    printf('Subscription created with exactly once delivery status: %s' . PHP_EOL, $status ? 'true' : 'false');
}

Surveiller les abonnements avec diffusion de type "exactement une fois"

La métrique subscription/exactly_once_warning_count enregistre le nombre d'événements pouvant entraîner une nouvelle distribution (valides ou en double). Cette métrique comptabilise le nombre de fois où Pub/Sub n'a pas réussi à traiter les requêtes associées aux ID d'accusé de réception (requête ModifyAckDeadline ou acknowledgment). Les raisons de l'échec peuvent être liées au serveur ou au client. Par exemple, si la couche de persistance utilisée pour conserver les informations de remise "exactement une fois" n'est pas disponible, il s'agit d'un événement côté serveur. Si le client tente d'accuser réception d'un message avec un ID d'accusé de réception non valide, il s'agit d'un événement basé sur le client.

Comprendre la métrique

subscription/exactly_once_warning_count capture les événements qui peuvent ou non entraîner des nouvelles livraisons réelles et peut être bruyant en fonction du comportement du client. Par exemple, les requêtes acknowledgment ou ModifyAckDeadline répétées avec des ID d'accusé de réception non valides incrémentent la métrique de manière répétée.

Les métriques suivantes sont également utiles pour comprendre le comportement du client :

• La métrique subscription/expired_ack_deadlines_count indique le nombre d'ID d'accusé de réception expirés. L'expiration des ID d'accusé de réception peut entraîner des échecs pour les requêtes ModifyAckDeadline et acknowledgment.

• La métrique service.serviceruntime.googleapis.com/api/request_count peut être utilisée pour enregistrer les échecs des requêtes ModifyAckDeadline ou acknowledgment dans les cas où les requêtes atteignent Trusted Cloud by S3NS , mais pas Pub/Sub. Cette métrique ne capture pas certaines défaillances, par exemple lorsque les clients sont déconnectés de Trusted Cloud by S3NS.

Dans la plupart des cas d'événements d'échec pouvant être réessayés, les bibliothèques clientes compatibles relancent automatiquement la requête.

Quotas

Les abonnements avec distribution de type "exactement une fois" sont soumis à des exigences de quota supplémentaires. Ces quotas sont appliqués aux éléments suivants :

• Nombre de messages consommés à partir d'abonnements avec distribution de type "exactement une fois" activée par région.
• Nombre de messages pour lesquels l'accusé de réception a été envoyé ou dont le délai a été prolongé lorsque des abonnements avec distribution de type "exactement une fois" sont activés par région.

Pour en savoir plus sur ces quotas, consultez le tableau de la section Quotas.

Distribution de type "exactement une fois" et abonnements ordonnés

Pub/Sub est compatible avec la distribution de type "exactement une fois" avec la distribution ordonnée.

Lorsque vous utilisez le tri avec la distribution de type "exactement une fois", Pub/Sub s'attend à ce que les accusés de réception soient dans l'ordre. Si les accusés de réception sont dans le désordre, le service échoue les requêtes avec des erreurs temporaires. Si le délai de confirmation expire avant la réception d'un accusé de réception dans l'ordre pour la remise, le client recevra une nouvelle remise du message. Par conséquent, lorsque vous utilisez le tri avec la distribution exactement une fois, le débit du client est limité à un ordre de grandeur de milliers de messages par seconde.

Distribution de type "exactement une fois" et abonnements push

Pub/Sub n'est compatible avec la distribution de type "exactement une fois" qu'avec les abonnements pull.

Les clients qui consomment des messages provenant d'abonnements push accusent réception des messages en répondant aux requêtes push par une réponse de réussite. Toutefois, les clients ne savent pas si l'abonnement Pub/Sub a reçu la réponse et l'a traitée. Cela diffère des abonnements par extraction, où les demandes d'accusé de réception sont initiées par les clients et l'abonnement Pub/Sub répond si la demande a été traitée avec succès. Pour cette raison, la sémantique de distribution de type "exactement une fois" ne s'aligne pas bien sur les abonnements push.

Bon à savoir

• Si le délai d'accusé de réception n'est pas spécifié lors de la création de l'abonnement, les abonnements pour lesquels la diffusion de type "exactement une fois" est activée auront un délai d'accusé de réception par défaut de 60 secondes.

• Des délais d'accusé de réception par défaut plus longs permettent d'éviter les nouvelles distributions causées par des événements réseau. Les bibliothèques clientes compatibles n'utilisent pas le délai de confirmation par défaut des abonnements.

• Les abonnements avec distribution de type "exactement une fois" présentent une latence de publication vers l'abonnement beaucoup plus élevée que les abonnements standards.

• Si vous avez besoin d'un débit élevé, vos clients de type "exactement une fois" doivent également utiliser l'extraction en flux continu.

• Un abonnement peut recevoir plusieurs copies du même message en raison de doublons côté publication, même si la diffusion de type "exactement une fois" est activée. Les doublons côté publication peuvent être dus à plusieurs nouvelles tentatives de publication uniques par le client de publication ou le service Pub/Sub. Plusieurs publications uniques par le client de publication, lors de nouvelles tentatives, entraînent des nouvelles remises avec des ID de message différents. Plusieurs publications uniques par le service Pub/Sub, en réponse à une demande de publication d'un client, entraînent des redistributions avec les mêmes ID de message.

• Vous pouvez réessayer en cas d'échec dans subscription/exactly_once_warning_count. Les bibliothèques clientes compatibles réessaient automatiquement. Toutefois, les échecs liés à des ID d'accusé de réception non valides ne peuvent pas être relancés.