Como receber mensagens de assinaturas do Lite

Esta página explica como receber mensagens de assinaturas Lite. Você pode receber mensagens com a biblioteca de cliente do Pub/Sub Lite para Java.

As assinaturas Lite conectam os tópicos Lite aos aplicativos dos assinantes. Os assinantes recebem mensagens de assinaturas Lite. Os assinantes recebem todas as mensagens que os aplicativos do editor enviam para o tópico do Lite, incluindo as mensagens que os editores enviam antes de criar a assinatura do Lite.

Antes de receber mensagens de uma assinatura do Lite, crie um tópico do Lite, uma assinatura do Lite para esse tópico e publique mensagens para o tópico do Lite.

Receber mensagens

Para receber mensagens de uma assinatura do Lite, solicite mensagens da assinatura do Lite. A biblioteca de cliente se conecta automaticamente às partições no tópico do Lite anexado à assinatura do Lite. Se mais de um cliente assinante for instanciado, as mensagens serão distribuídas entre todos os clientes. O número de partições no tópico determina o número máximo de clientes assinantes que podem se conectar simultaneamente a uma assinatura.

O exemplo a seguir mostra como receber mensagens de assinaturas do Lite:

Go

Antes de tentar esse exemplo, siga as instruções de configuração do Go em Guia de início rápido: como usar bibliotecas de cliente. Para mais informações, consulte a documentação de referência da API Pub/Sub Go.

Veja no GitHub Feedback

package main

import (
	"context"
	"flag"
	"fmt"
	"log"
	"sync/atomic"
	"time"

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

func main() {
	// NOTE: Set these flags for an existing Pub/Sub Lite subscription containing
	// published messages when running this sample.
	projectID := flag.String("project_id", "", "Cloud Project ID")
	zone := flag.String("zone", "", "Cloud Zone where the topic resides, e.g. us-central1-a")
	subscriptionID := flag.String("subscription_id", "", "Existing Pub/Sub Lite subscription")
	timeout := flag.Duration("timeout", 90*time.Second, "The duration to receive messages")
	flag.Parse()

	ctx := context.Background()
	subscriptionPath := fmt.Sprintf("projects/%s/locations/%s/subscriptions/%s", *projectID, *zone, *subscriptionID)

	// Configure flow control settings. These settings apply per partition.
	// The message stream is paused based on the maximum size or number of
	// messages that the subscriber has already received, whichever condition is
	// met first.
	settings := pscompat.ReceiveSettings{
		// 10 MiB. Must be greater than the allowed size of the largest message
		// (1 MiB).
		MaxOutstandingBytes: 10 * 1024 * 1024,
		// 1,000 outstanding messages. Must be > 0.
		MaxOutstandingMessages: 1000,
	}

	// Create the subscriber client.
	subscriber, err := pscompat.NewSubscriberClientWithSettings(ctx, subscriptionPath, settings)
	if err != nil {
		log.Fatalf("pscompat.NewSubscriberClientWithSettings error: %v", err)
	}

	// Listen for messages until the timeout expires.
	log.Printf("Listening to messages on %s for %v...\n", subscriptionPath, *timeout)
	cctx, cancel := context.WithTimeout(ctx, *timeout)
	defer cancel()
	var receiveCount int32

	// Receive blocks until the context is cancelled or an error occurs.
	if err := subscriber.Receive(cctx, func(ctx context.Context, msg *pubsub.Message) {
		// NOTE: May be called concurrently; synchronize access to shared memory.
		atomic.AddInt32(&receiveCount, 1)

		// Metadata decoded from the message ID contains the partition and offset.
		metadata, err := pscompat.ParseMessageMetadata(msg.ID)
		if err != nil {
			log.Fatalf("Failed to parse %q: %v", msg.ID, err)
		}

		fmt.Printf("Received (partition=%d, offset=%d): %s\n", metadata.Partition, metadata.Offset, string(msg.Data))
		msg.Ack()
	}); err != nil {
		log.Fatalf("SubscriberClient.Receive error: %v", err)
	}

	fmt.Printf("Received %d messages\n", receiveCount)
}

Java

Antes de executar este exemplo, siga as instruções de configuração do Java nas bibliotecas de cliente do Pub/Sub Lite.

Ver no GitHub Feedback

import com.google.api.gax.rpc.ApiException;
import com.google.cloud.pubsub.v1.AckReplyConsumer;
import com.google.cloud.pubsub.v1.MessageReceiver;
import com.google.cloud.pubsublite.CloudRegion;
import com.google.cloud.pubsublite.CloudZone;
import com.google.cloud.pubsublite.MessageMetadata;
import com.google.cloud.pubsublite.ProjectNumber;
import com.google.cloud.pubsublite.SubscriptionName;
import com.google.cloud.pubsublite.SubscriptionPath;
import com.google.cloud.pubsublite.cloudpubsub.FlowControlSettings;
import com.google.cloud.pubsublite.cloudpubsub.Subscriber;
import com.google.cloud.pubsublite.cloudpubsub.SubscriberSettings;
import com.google.pubsub.v1.PubsubMessage;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

public class SubscriberExample {

  public static void main(String... args) throws Exception {
    // TODO(developer): Replace these variables before running the sample.
    String cloudRegion = "your-cloud-region";
    char zoneId = 'b';
    // Choose an existing subscription for the subscribe example to work.
    String subscriptionId = "your-subscription-id";
    long projectNumber = Long.parseLong("123456789");

    subscriberExample(cloudRegion, zoneId, projectNumber, subscriptionId);
  }

  public static void subscriberExample(
      String cloudRegion, char zoneId, long projectNumber, String subscriptionId)
      throws ApiException {

    SubscriptionPath subscriptionPath =
        SubscriptionPath.newBuilder()
            .setLocation(CloudZone.of(CloudRegion.of(cloudRegion), zoneId))
            .setProject(ProjectNumber.of(projectNumber))
            .setName(SubscriptionName.of(subscriptionId))
            .build();

    // The message stream is paused based on the maximum size or number of messages that the
    // subscriber has already received, whichever condition is met first.
    FlowControlSettings flowControlSettings =
        FlowControlSettings.builder()
            // 10 MiB. Must be greater than the allowed size of the largest message (1 MiB).
            .setBytesOutstanding(10 * 1024 * 1024L)
            // 1,000 outstanding messages. Must be >0.
            .setMessagesOutstanding(1000L)
            .build();

    MessageReceiver receiver =
        (PubsubMessage message, AckReplyConsumer consumer) -> {
          System.out.println("Id : " + MessageMetadata.decode(message.getMessageId()));
          System.out.println("Data : " + message.getData().toStringUtf8());
          consumer.ack();
        };

    SubscriberSettings subscriberSettings =
        SubscriberSettings.newBuilder()
            .setSubscriptionPath(subscriptionPath)
            .setReceiver(receiver)
            // Flow control settings are set at the partition level.
            .setPerPartitionFlowControlSettings(flowControlSettings)
            .build();

    Subscriber subscriber = Subscriber.create(subscriberSettings);

    // Start the subscriber. Upon successful starting, its state will become RUNNING.
    subscriber.startAsync().awaitRunning();

    System.out.println("Listening to messages on " + subscriptionPath.toString() + "...");

    try {
      System.out.println(subscriber.state());
      // Wait 90 seconds for the subscriber to reach TERMINATED state. If it encounters
      // unrecoverable errors before then, its state will change to FAILED and an
      // IllegalStateException will be thrown.
      subscriber.awaitTerminated(90, TimeUnit.SECONDS);
    } catch (TimeoutException t) {
      // Shut down the subscriber. This will change the state of the subscriber to TERMINATED.
      subscriber.stopAsync().awaitTerminated();
      System.out.println("Subscriber is shut down: " + subscriber.state());
    }
  }
}

Python

Antes de executar este exemplo, siga as instruções de configuração do Java nas bibliotecas de cliente do Pub/Sub Lite.

Ver no GitHub Feedback

from concurrent.futures._base import TimeoutError
from google.cloud.pubsublite.cloudpubsub import SubscriberClient
from google.cloud.pubsublite.types import (
    CloudRegion,
    CloudZone,
    FlowControlSettings,
    SubscriptionPath,
)

# TODO(developer):
# project_number = 1122334455
# cloud_region = "us-central1"
# zone_id = "a"
# subscription_id = "your-subscription-id"
# timeout = 90

location = CloudZone(CloudRegion(cloud_region), zone_id)
subscription_path = SubscriptionPath(project_number, location, subscription_id)
# Configure when to pause the message stream for more incoming messages based on the
# maximum size or number of messages that a single-partition subscriber has received,
# whichever condition is met first.
per_partition_flow_control_settings = FlowControlSettings(
    # 1,000 outstanding messages. Must be >0.
    messages_outstanding=1000,
    # 10 MiB. Must be greater than the allowed size of the largest message (1 MiB).
    bytes_outstanding=10 * 1024 * 1024,
)

def callback(message: PubsubMessage):
    message_data = message.data.decode("utf-8")
    metadata = MessageMetadata.decode(message.message_id)
    print(f"Received {message_data} of ordering key {message.ordering_key} with id {metadata}.")
    message.ack()

# SubscriberClient() must be used in a `with` block or have __enter__() called before use.
with SubscriberClient() as subscriber_client:

    streaming_pull_future = subscriber_client.subscribe(
        subscription_path,
        callback=callback,
        per_partition_flow_control_settings=per_partition_flow_control_settings,
    )

    print(f"Listening for messages on {str(subscription_path)}...")

    try:
        streaming_pull_future.result(timeout=timeout)
    except TimeoutError or KeyboardInterrupt:
        streaming_pull_future.cancel()
        assert streaming_pull_future.done()

A biblioteca de cliente estabelece conexões de streaming bidirecionais para cada uma das partições no tópico do Lite.

1. O assinante solicita conexões às partições.

2. O serviço Pub/Sub Lite entrega as mensagens ao assinante.

Depois de processar a mensagem, o assinante precisa confirmá-la. A biblioteca de cliente processa e confirma mensagens de maneira assíncrona em um callback. Para limitar o número de mensagens não confirmadas que o assinante pode armazenar na memória, defina as configurações de controle de fluxo.

Se vários assinantes receberem mensagens da mesma assinatura do Lite, o serviço Lite do Pub/Sub conectará cada assinante a uma proporção igual de partições. Por exemplo, se dois assinantes usarem a mesma assinatura do Lite e a assinatura do Lite estiver anexada a um tópico do Lite com duas partições, cada assinante receberá mensagens de uma das partições.

Como confirmar mensagens

Para confirmar uma mensagem, envie uma confirmação para a assinatura Lite.

Java

Para enviar uma confirmação, use o método AckReplyConsumer.ack().

Cuidado: o Pub/Sub Lite não é compatível com confirmações negativas. Se você chamar o método AckReplyConsumer.nack(), a biblioteca de cliente fechará as conexões de streaming e será necessário solicitar as mensagens novamente.

Python

Para enviar uma confirmação, use o método Message.ack().

Cuidado: o Pub/Sub Lite não é compatível com confirmações negativas. Se você chamar o método Message.nack(), a biblioteca de cliente fechará as conexões de streaming e será necessário solicitar as mensagens novamente.

Os assinantes precisam confirmar todas as mensagens. Os assinantes recebem a mensagem não confirmada mais antiga primeiro, seguida de cada mensagem subsequente. Se um assinante pular uma mensagem, confirmar as próximas mensagens e se reconectar, o assinante receberá a mensagem não confirmada e cada mensagem subsequente e confirmada.

As assinaturas do Lite não têm um prazo de confirmação e o serviço Pub/Sub Lite não reenvia mensagens não confirmadas por uma conexão de streaming aberto.

Como usar o controle de fluxo

Depois que o serviço do Pub/Sub Lite entrega mensagens aos assinantes, eles armazenam mensagens não confirmadas na memória. É possível limitar o número de mensagens pendentes que os assinantes podem armazenar na memória usando as configurações de controle de fluxo. As configurações de controle de fluxo são aplicadas a cada partição de que um assinante recebe mensagens.

É possível definir as seguintes configurações de controle de fluxo:

• Tamanho de mensagem pendente. O tamanho máximo, em bytes, das mensagens subsequentes. O tamanho máximo precisa ser maior que o tamanho da maior mensagem.
• Número de mensagens. O número máximo de mensagens pendentes.

O tamanho de uma mensagem está no campo size_bytes (em inglês). É possível definir as configurações de controle de fluxo com a biblioteca de cliente.

Java

Para definir as configurações do controle de fluxo, use os seguintes métodos na classe FlowControlRequest.Builder:

• setBytesOutstanding

• setMessagesOutstanding

Python

Para definir as configurações do controle de fluxo, defina os seguintes parâmetros na classe FlowControlSettings:

• bytes_outstanding

• messages_outstanding

Por exemplo, se o número máximo de mensagens for 100 e o assinante se conectar a 10 partições, não será possível receber mais de 100 mensagens de nenhuma das 10 partições. O número total de mensagens pendentes pode ser maior que 100, mas o assinante não pode armazenar mais de 100 mensagens de cada partição.