Ricezione di messaggi da sottoscrizioni Lite

Nota: Pub/Sub Lite è deprecato. A partire dal 18 marzo 2026, Pub/Sub Lite verrà disattivato.

• Clienti attuali: Pub/Sub Lite rimarrà funzionale fino al 18 marzo 2026. Se non hai utilizzato Pub/Sub Lite nel periodo di 90 giorni precedente al 24 settembre 2024 (26 giugno 2024 - 24 settembre 2024), non potrai accedere a Pub/Sub Lite a partire dal 24 settembre 2024.
• Nuovi clienti: Pub/Sub Lite non è più disponibile per i nuovi clienti dopo il 24 settembre 2024.

Puoi eseguire la migrazione del tuo servizio Pub/Sub Lite a Google Cloud Managed Service per Apache Kafka oppure a Pub/Sub.

Questa pagina spiega come ricevere messaggi dagli abbonamenti Lite. Puoi ricevere messaggi con la libreria client Pub/Sub Lite per Java.

Le sottoscrizioni Lite collegano gli argomenti Lite alle applicazioni dei sottoscrittori, che ricevono i messaggi dalle sottoscrizioni Lite. Gli abbonati ricevono tutti i messaggi inviati dalle applicazioni del publisher all'argomento Lite, inclusi i messaggi inviati dai publisher prima della creazione dell'abbonamento Lite.

Prima di ricevere messaggi da una sottoscrizione Lite, crea un argomento Lite, crea una sottoscrizione Lite all'argomento Lite e pubblica messaggi nell'argomento Lite.

Ricezione di messaggi

Per ricevere messaggi da un abbonamento Lite, richiedi i messaggi dall'abbonamento Lite. La libreria client si connette automaticamente partizioni nell'argomento Lite collegato alla sottoscrizione Lite. Se il client di più sottoscrittori Creata un'istanza, i messaggi verranno distribuiti tra tutti i client. Il numero di partizioni nell'argomento determina il numero massimo di client di abbonati che possono connettersi contemporaneamente a una sottoscrizione.

L'inizializzazione e l'inizio della ricezione da parte degli sottoscrittori potrebbero richiedere fino a un minuto messaggi. Dopo l'inizializzazione, i messaggi vengono ricevuti con una latenza minima.

L'esempio seguente mostra come ricevere messaggi dalle sottoscrizioni Lite:

gcloud

Questo comando richiede Python 3.6 o versioni successive e l'installazione del pacchetto Python grpcio. Per gli utenti di MacOS, Linux e Cloud Shell, esegui:

sudo pip3 install grpcio
export CLOUDSDK_PYTHON_SITEPACKAGES=1

Per ricevere messaggi, utilizza gcloud pubsub lite-subscriptions Subscribe :

gcloud pubsub lite-subscriptions subscribe SUBSCRIPTION_ID \
    --location=LITE_LOCATION \
    --auto-ack

Sostituisci quanto segue:

• SUBSCRIPTION_ID: l'ID dell'abbonamento Lite
• LITE_LOCATION: la località della sottoscrizione Lite

Vai

Prima di eseguire questo esempio, segui le istruzioni di configurazione di Go in Librerie client Pub/Sub Lite.

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

Prima di eseguire questo esempio, segui le istruzioni di configurazione di Java riportate in Librerie client Pub/Sub Lite.

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.CloudRegionOrZone;
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.MessageTransforms;
import com.google.cloud.pubsublite.cloudpubsub.Subscriber;
import com.google.cloud.pubsublite.cloudpubsub.SubscriberSettings;
import com.google.protobuf.Timestamp;
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");
    // True if using a regional location. False if using a zonal location.
    // https://cloud.google.com/pubsub/lite/docs/topics
    boolean regional = false;

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

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

    CloudRegionOrZone location;
    if (regional) {
      location = CloudRegionOrZone.of(CloudRegion.of(cloudRegion));
    } else {
      location = CloudRegionOrZone.of(CloudZone.of(CloudRegion.of(cloudRegion), zoneId));
    }

    SubscriptionPath subscriptionPath =
        SubscriptionPath.newBuilder()
            .setLocation(location)
            .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());
          System.out.println("Ordering key : " + message.getOrderingKey());
          System.out.println("Attributes : ");
          message
              .getAttributesMap()
              .forEach(
                  (key, value) -> {
                    if (key == MessageTransforms.PUBSUB_LITE_EVENT_TIME_TIMESTAMP_PROTO) {
                      Timestamp ts = MessageTransforms.decodeAttributeEventTime(value);
                      System.out.println(key + " = " + ts.toString());
                    } else {
                      System.out.println(key + " = " + value);
                    }
                  });

          // Acknowledge the message.
          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

Prima di eseguire questo esempio, segui le istruzioni di configurazione di Python in Librerie client Pub/Sub Lite.

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

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

if regional:
    location = CloudRegion(cloud_region)
else:
    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()

La libreria client stabilisce connessioni di flusso bidirezionali tra delle partizioni nell'argomento Lite.

1. L'abbonato richiede connessioni alle partizioni.

2. Il servizio Pub/Sub Lite consegna i messaggi al sottoscrittore.

Dopo aver elaborato il messaggio, l'abbonato deve riconoscerlo. La libreria client elabora e conferma i messaggi in modo asincrono in un callback. Per limitare il numero di messaggi non confermati che l'abbonato può memorizzare in memoria, configura le impostazioni di controllo del flusso.

Se più sottoscrittori ricevono messaggi dalla stessa sottoscrizione Lite, Il servizio Pub/Sub Lite connette ogni sottoscrittore in egual misura delle partizioni. Ad esempio, se due sottoscrittori usano la stessa sottoscrizione Lite e la sottoscrizione Lite è collegata a un argomento Lite con due partizioni, il sottoscrittore riceve messaggi da una delle partizioni.

Confermare i messaggi

Per confermare un messaggio, invia una conferma alla sottoscrizione Lite.

Vai

Per inviare una conferma, utilizza il metodo Message.Ack().

Attenzione: Pub/Sub Lite non supporta conferme negative. Per impostazione predefinita, viene terminato il SubscriberClient. È possibile fornire una funzione personalizzata per ReceiveSettings.NackHandler per gestire i messaggi bloccati.

Java

Per inviare un messaggio di conferma, utilizza il metodo AckReplyConsumer.ack().

Attenzione: Pub/Sub Lite non supporta gli accoglimenti negativi. Se chiami il metodo AckReplyConsumer.nack(), la libreria client si chiude connessioni di streaming esistenti ed è necessario richiedere di nuovo i messaggi.

Python

Per inviare una conferma, utilizza il metodo Message.ack().

Attenzione: Pub/Sub Lite non supporta conferme negative. Se chiami il metodo Message.nack(), la libreria client si chiude connessioni di streaming esistenti ed è necessario richiedere di nuovo i messaggi.

I sottoscrittori devono confermare ogni messaggio. Gli abbonati ricevono prima il messaggio senza ACK meno recente, seguito da ogni messaggio successivo. Se Il sottoscrittore ignora un messaggio, conferma i messaggi successivi e si riconnette, il sottoscrittore riceve il messaggio non confermato e ogni successivo messaggio confermato.

Le sottoscrizioni Lite non hanno una scadenza per la conferma di ricezione e il servizio Pub/Sub Lite non recapita i messaggi non confermati tramite una connessione di streaming aperta.

Utilizzo del controllo del flusso

Una volta che il servizio Pub/Sub Lite consegna i messaggi ai sottoscrittori, I sottoscrittori archiviano messaggi non confermati in memoria. Puoi limitare il numero di messaggi in sospeso che gli abbonati possono memorizzare in memoria utilizzando le impostazioni di controllo del flusso. Le impostazioni di controllo del flusso si applicano a ogni partizione da cui un abbonato riceve i messaggi.

Puoi configurare le seguenti impostazioni di controllo flusso:

• Dimensioni dei messaggi in attesa. La dimensione massima, in byte, messaggi in sospeso. La dimensione massima deve essere maggiore della dimensione del messaggio più grande.
• Numero di messaggi. Il numero massimo di messaggi in sospeso.

Le dimensioni di un messaggio si trovano nel campo size_bytes. Puoi configurare le impostazioni di controllo del flusso con la libreria client.

Vai

Per configurare le impostazioni di controllo del flusso, inserisci ReceiveSettings durante la chiamata a pscompat.NewSubscriberClientWithSettings. In ReceiveSettings puoi impostare i seguenti parametri:

• MaxOutstandingMessages

• MaxOutstandingBytes

Per un esempio, consulta questo esempio di controllo del flusso.

Java

Per configurare le impostazioni di controllo del flusso, utilizza i seguenti metodi nella classe FlowControlRequest.Builder:

• setBytesOutstanding

• setMessagesOutstanding

Python

Per configurare le impostazioni di controllo flusso, imposta i seguenti parametri nella classe FlowControlSettings:

• bytes_outstanding

• messages_outstanding

Ad esempio, se il numero massimo di messaggi è 100 e il sottoscrittore si connette a 10 partizioni, il sottoscrittore non può ricevere più di 100 messaggi da una qualsiasi delle 10 partizioni. Il numero totale di messaggi in sospeso potrebbe essere maggiore di 100, ma il sottoscrittore non può archiviare più di 100 messaggi provenienti da ogni partizione.