Ordenar mensagens

A ordenação de mensagens é um recurso do Pub/Sub que permite receber mensagens nos clientes assinantes na ordem em que foram publicadas pelos clientes editores.

Por exemplo, suponha que um cliente editor em uma região publique as mensagens 1, 2 e 3 em ordem. Com a ordenação de mensagens, o cliente assinante recebe as mensagens publicadas na mesma ordem. Para serem entregues em ordem, o cliente editor precisa publicar as mensagens na mesma região. No entanto, os assinantes podem se conectar a qualquer região, e a garantia de ordenação ainda é mantida.

A ordenação de mensagens é um recurso útil para cenários como captura de mudanças no banco de dados, rastreamento de sessões de usuários e aplicativos de streaming em que é importante preservar a cronologia dos eventos.

Esta página explica o conceito de ordenação de mensagens e como configurar os clientes assinantes para receber mensagens em ordem. Para configurar os clientes de publicação para ordenação de mensagens, consulte Usar chaves de ordenação para publicar uma mensagem.

Visão geral da ordem das mensagens

A ordenação no Pub/Sub é determinada pelo seguinte:

• Chave de ordenação. Uma chave de ordenação é uma string que identifica mensagens relacionadas que precisam ser ordenadas. Exemplos de chaves de ordenação incluem IDs de clientes ou a chave primária de uma linha em um banco de dados. Uma chave de ordenação pode ter até 1 KB de comprimento.

  Para conseguir a ordenação de mensagens, defina a mesma chave em todas as mensagens relacionadas que precisam ser recebidas em ordem. Além disso, é necessário publicar todas as mensagens com a mesma chave de ordenação na mesma região. Para mais informações, consulte Usar chaves de ordenação para publicar uma mensagem.

  As mensagens com uma chave de ordenação vazia não são ordenadas.

• Ative a ordenação de mensagens. Para receber as mensagens em ordem, ative a ordenação de mensagens na assinatura. Para mais informações, consulte Ativar a ordenação de mensagens.

  Se a ordenação de mensagens não estiver ativada, uma assinatura vai receber mensagens sem uma ordem esperada. Por exemplo, suponha que as assinaturas A e B estejam anexadas ao mesmo tópico T e que a ordenação esteja ativada na assinatura A, mas não na B. Embora as duas assinaturas recebam o mesmo conjunto de mensagens do tópico T, a ordenação só é preservada para a assinatura A.

A taxa de transferência de publicação em cada chave de ordenação é limitada a 1 MBps. O throughput em todas as chaves de ordenação de um tópico é limitado à cota disponível em uma região de publicação. Esse limite pode ser aumentado para vários GBps.

Em geral, se a solução exigir que os clientes publishers enviem mensagens ordenadas e não ordenadas, crie tópicos separados, um para mensagens ordenadas e outro para mensagens não ordenadas.

Considerações ao usar mensagens ordenadas

A lista a seguir contém informações importantes sobre o comportamento do envio de mensagens ordenadas no Pub/Sub:

• Cardinalidade. Uma chave de ordenação não é equivalente a uma partição em um sistema de mensagens baseado em partições, porque as chaves de ordenação têm uma cardinalidade muito maior do que as partições.

• Ordenação dentro da chave: as mensagens publicadas com a mesma chave de ordenação devem ser recebidas em ordem. Suponha que, para a chave de ordenação A, você publique as mensagens 1, 2 e 3. Com a ordenação ativada, espera-se que 1 seja entregue antes de 2 e 2 antes de 3.

• Ordenação entre chaves: não é esperado que as mensagens publicadas com chaves de ordenação diferentes sejam recebidas em ordem. Suponha que você tenha chaves de ordenação A e B. Para a chave de ordenação A, as mensagens 1 e 2 são publicadas em ordem. Para a chave de ordenação B, as mensagens 3 e 4 são publicadas em ordem. No entanto, a mensagem 1 pode chegar antes ou depois da mensagem 4.

• Reenvio de mensagens: o Pub/Sub entrega cada mensagem pelo menos uma vez. Portanto, o serviço do Pub/Sub pode reenviar mensagens. Os reenvios de uma mensagem acionam o reenvio de todas as mensagens subsequentes para essa chave, mesmo as confirmadas. Suponha que um cliente assinante receba as mensagens 1, 2 e 3 para uma chave de ordenação específica. Se a mensagem 2 for reenviada (porque o prazo de confirmação expirou ou a confirmação de melhor esforço não persistiu no Pub/Sub), a mensagem 3 também será reenviada. Se a ordem das mensagens e um tópico de mensagens inativas estiverem ativados em uma assinatura, esse comportamento poderá não ser verdadeiro, já que o Pub/Sub encaminha mensagens para tópicos de mensagens inativas com base no melhor esforço.

• Atrasos no reconhecimento e tópicos de mensagens inativas: mensagens não reconhecidas para uma determinada chave de ordenação podem atrasar a entrega de mensagens para outras chaves, principalmente durante reinicializações do servidor ou mudanças no tráfego. Para manter a ordem nesses eventos, confirme o recebimento de todas as mensagens a tempo. Se não for possível fazer o reconhecimento em tempo hábil, use um tópico de mensagens inativas para evitar a retenção indefinida de mensagens. A ordem pode não ser preservada quando as mensagens são gravadas em um tópico de mensagens inativas.

• Afinidade de mensagens (clientes streamingPull): as mensagens com a mesma chave geralmente são entregues ao mesmo cliente assinante streamingPull. A afinidade é esperada quando há mensagens pendentes para uma chave de ordenação de um cliente assinante específico. Se não houver mensagens pendentes, a afinidade poderá mudar para balanceamento de carga ou desconexões do cliente.

  Para garantir um processamento tranquilo mesmo com possíveis mudanças de afinidade, é crucial projetar seu aplicativo streamingPull de forma que ele possa processar mensagens em qualquer cliente para uma determinada chave de ordenação.

• Integração com o Dataflow: não ative a ordenação de mensagens para assinaturas ao configurar o Dataflow com o Pub/Sub. O Dataflow tem um mecanismo próprio para ordenação total de mensagens, garantindo a ordem cronológica em todas as mensagens como parte das operações de janela. Esse método de ordenação difere da abordagem baseada em chaves de ordenação do Pub/Sub. O uso de chaves de ordenação com o Dataflow pode reduzir a performance do pipeline.

• Escalonamento automático: a entrega ordenada do Pub/Sub é escalonada para bilhões de chaves de ordenação. Um número maior de chaves de ordenação permite mais entregas paralelas aos assinantes, já que a ordenação se aplica a todas as mensagens com a mesma chave de ordenação.

• Compensações de performance: a entrega ordenada tem algumas compensações. Em comparação com a entrega não ordenada, a entrega ordenada diminui a disponibilidade de publicação e aumenta a latência de entrega de mensagens de ponta a ponta. No caso de entrega ordenada, o failover exige coordenação para garantir que as mensagens sejam gravadas e lidas na ordem correta.

• Tecla de atalho: ao usar a ordenação de mensagens, todas as mensagens com a mesma chave de ordenação são enviadas ao cliente assinante na ordem em que são recebidas pelo serviço. O callback do usuário não é executado até que o callback seja concluído para a mensagem anterior. A capacidade de processamento máxima para mensagens que compartilham a mesma chave de ordenação ao entregar para assinantes não é limitada pelo Pub/Sub , mas pela velocidade de processamento do cliente assinante. Uma tecla de atalho ocorre quando um backlog é criado em uma chave de ordem individual porque o número de mensagens produzidas por segundo excede o número de mensagens que o assinante pode processar por segundo. Para reduzir as teclas de atalho, use as teclas mais granulares possíveis e minimize o tempo de processamento por mensagem. Também é possível monitorar a métrica subscription/oldest_unacked_message_age para um valor crescente, o que pode indicar uma tecla de atalho.

Para mais informações sobre como usar a ordenação de mensagens, consulte os seguintes tópicos de práticas recomendadas:

• Práticas recomendadas para mensagens ordenadas na publicação

• Práticas recomendadas para mensagens ordenadas na inscrição

Comportamento do cliente assinante para ordenação de mensagens

Os clientes assinantes recebem mensagens na ordem em que foram publicadas em uma região específica. O Pub/Sub oferece suporte a diferentes maneiras de receber mensagens, como clientes assinantes conectados a assinaturas de pull e push. As bibliotecas de cliente usam streamingPull (com exceção do PHP).

Para saber mais sobre esses tipos de assinatura, consulte Escolher um tipo de assinatura.

As seções a seguir explicam o que significa receber mensagens em ordem para cada tipo de cliente assinante.

Clientes assinantes do StreamingPull

Ao usar as bibliotecas de cliente com streamingPull, é necessário especificar um callback de usuário que é executado sempre que uma mensagem é recebida por um cliente assinante. Com as bibliotecas de cliente, para qualquer chave de ordenação, o callback é executado até a conclusão nas mensagens na ordem correta. Se as mensagens forem confirmadas nesse callback, todas as computações em uma mensagem ocorrerão em ordem. No entanto, se o callback do usuário agendar outro trabalho assíncrono em mensagens, o cliente assinante precisará garantir que o trabalho assíncrono seja feito em ordem. Uma opção é adicionar mensagens a uma fila de trabalho local que é processada em ordem.

Clientes de assinantes por pull

Para clientes assinantes conectados a assinaturas de pull, a ordenação de mensagens do Pub/Sub oferece suporte ao seguinte:

• Todas as mensagens de uma chave de ordenação na PullResponse estão na ordem correta na lista.

• Apenas um lote de mensagens pode ficar pendente para uma chave de ordenação por vez.

O requisito de que apenas um lote de mensagens possa estar pendente por vez é necessário para manter a entrega ordenada, já que o serviço do Pub/Sub não pode garantir o sucesso ou a latência da resposta enviada para uma solicitação de envio de um assinante.

Clientes de assinantes de push

As restrições de push são ainda mais rigorosas do que as de pull. Para uma assinatura por push, o Pub/Sub aceita apenas uma mensagem pendente por chave de ordenação por vez. Cada mensagem é enviada para um endpoint de push como uma solicitação separada. Portanto, enviar as solicitações em paralelo teria o mesmo problema que entregar vários lotes de mensagens para a mesma chave de ordenação e extrair assinantes simultaneamente. As assinaturas por push podem não ser uma boa opção para tópicos em que as mensagens são publicadas com frequência com a mesma chave de ordenação ou em que a latência é extremamente importante.

Exportar clientes assinantes

As assinaturas de exportação aceitam mensagens ordenadas. Para assinaturas do BigQuery, as mensagens com a mesma chave de ordenação são gravadas na tabela do BigQuery em ordem. Para assinaturas do Cloud Storage, nem todas as mensagens com a mesma chave de ordenação podem ser gravadas no mesmo arquivo. Quando estão no mesmo arquivo, as mensagens de uma chave de ordenação ficam em ordem. Quando distribuídas em vários arquivos, as mensagens posteriores de uma chave de ordenação podem aparecer em um arquivo com um carimbo de data/hora anterior ao carimbo no nome do arquivo com as mensagens anteriores.

Ativar a ordenação de mensagens

Para receber as mensagens em ordem, defina a propriedade de ordenação das mensagens na assinatura que recebe as mensagens. O recebimento de mensagens pode aumentar a latência. Não é possível mudar a propriedade de ordenação de mensagens depois de criar uma assinatura.

É possível definir a propriedade de ordenação de mensagens ao criar uma assinatura usando o console Cloud de Confiance , a Google Cloud CLI ou a 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."

use google_cloud_pubsub::client::SubscriptionAdmin;

pub async fn sample(
    client: &SubscriptionAdmin,
    project_id: &str,
    topic_id: &str,
    subscription_id: &str,
) -> anyhow::Result<()> {
    let subscription_name = format!("projects/{project_id}/subscriptions/{subscription_id}");
    let topic_name = format!("projects/{project_id}/topics/{topic_id}");

    let subscription = client
        .create_subscription()
        .set_name(subscription_name)
        .set_topic(topic_name)
        .set_enable_message_ordering(true)
        .send()
        .await?;

    println!("successfully created subscription {subscription:?}");
    Ok(())
}

A seguir

• Leia a postagem do blog sobre entrega ordenada.

• Para continuar publicando mensagens em caso de erros não repetíveis, consulte Repetir solicitações com chaves de ordenação.

• Monitore sua assinatura.

• Leia sobre a publicação de mensagens com chaves de ordenação.