Menerima pesan dari langganan Lite

Catatan: Pub/Sub Lite tidak digunakan lagi. Mulai 18 Maret 2026, Pub/Sub Lite akan dinonaktifkan.

• Pelanggan saat ini: Pub/Sub Lite tetap berfungsi hingga 18 Maret 2026. Jika Anda belum menggunakan Pub/Sub Lite dalam periode 90 hari sebelum 24 September 2024 (26 Juni 2024 - 24 September 2024), Anda tidak akan dapat mengakses Pub/Sub Lite mulai 24 September 2024.
• Pelanggan baru: Pub/Sub Lite tidak lagi tersedia untuk pelanggan baru setelah 24 September 2024.

Anda dapat memigrasikan layanan Pub/Sub Lite ke Google Cloud Managed Service for Apache Kafka atau Pub/Sub.

Halaman ini menjelaskan cara menerima pesan dari langganan Lite. Anda dapat menerima pesan dengan library klien Pub/Sub Lite untuk Java.

Langganan Lite menghubungkan topik Lite ke aplikasi pelanggan; pelanggan menerima pesan dari langganan Lite. Pelanggan menerima setiap pesan yang dikirim aplikasi penayang ke topik Lite, termasuk pesan yang dikirim penayang sebelum Anda membuat langganan Lite.

Sebelum menerima pesan dari langganan Lite, buat topik Lite, buat langganan Lite ke topik Lite, dan publikasikan pesan ke topik Lite.

Menerima pesan

Untuk menerima pesan dari langganan Lite, minta pesan dari langganan Lite. Library klien akan otomatis terhubung ke partisi dalam topik Lite yang dilampirkan ke langganan Lite. Jika lebih dari satu klien subscriber dibuat instance-nya, pesan akan didistribusikan ke semua klien. Jumlah partisi dalam topik menentukan jumlah maksimum klien pelanggan yang dapat terhubung secara bersamaan ke langganan.

Pelanggan mungkin memerlukan waktu hingga satu menit untuk melakukan inisialisasi dan mulai menerima pesan. Setelah inisialisasi, pesan diterima dengan latensi minimal.

Contoh berikut menunjukkan cara menerima pesan dari langganan Lite:

gcloud

Perintah ini memerlukan Python 3.6 atau yang lebih baru, dan memerlukan paket Python gRPCio untuk diinstal. Untuk pengguna MacOS, Linux, dan Cloud Shell, jalankan:

sudo pip3 install grpcio
export CLOUDSDK_PYTHON_SITEPACKAGES=1

Untuk menerima pesan, gunakan perintah gcloud pubsub lite-subscriptions subscribe:

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

Ganti kode berikut:

• SUBSCRIPTION_ID: ID langganan Lite
• LITE_LOCATION: lokasi langganan Lite

Go

Sebelum menjalankan contoh ini, ikuti petunjuk penyiapan Go di Library Klien 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

Sebelum menjalankan contoh ini, ikuti petunjuk penyiapan Java di Library Klien 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

Sebelum menjalankan contoh ini, ikuti petunjuk penyiapan Python di Library Klien 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()

Library klien membuat koneksi streaming dua arah ke setiap partisi dalam topik Lite.

1. Pelanggan meminta koneksi ke partisi.

2. Layanan Pub/Sub Lite mengirimkan pesan ke pelanggan.

Setelah memproses pesan, pelanggan harus mengonfirmasi pesan. Library klien memproses dan mengonfirmasi pesan secara asinkron dalam callback. Untuk membatasi jumlah pesan yang tidak dikonfirmasi yang dapat disimpan pelanggan dalam memori, konfigurasi setelan kontrol alur.

Jika beberapa subscriber menerima pesan dari langganan Lite yang sama, layanan Pub/Sub Lite akan menghubungkan setiap subscriber ke proporsi partisi yang sama. Misalnya, jika dua pelanggan menggunakan langganan Lite yang sama dan langganan Lite dilampirkan ke topik Lite dengan dua partisi, setiap pelanggan akan menerima pesan dari salah satu partisi.

Mengonfirmasi pesan

Untuk mengonfirmasi pesan, kirim konfirmasi ke langganan Lite.

Go

Untuk mengirim konfirmasi, gunakan metode Message.Ack().

Perhatian: Pub/Sub Lite tidak mendukung konfirmasi negatif. Secara default, tindakan ini akan menghentikan SubscriberClient. Fungsi kustom dapat disediakan untuk ReceiveSettings.NackHandler guna menangani pesan yang ditolak.

Java

Untuk mengirim konfirmasi, gunakan metode AckReplyConsumer.ack().

Perhatian: Pub/Sub Lite tidak mendukung konfirmasi negatif. Jika Anda memanggil metode AckReplyConsumer.nack(), library klien akan menutup koneksi streaming yang ada, dan Anda harus meminta pesan lagi.

Python

Untuk mengirim konfirmasi, gunakan metode Message.ack().

Perhatian: Pub/Sub Lite tidak mendukung konfirmasi negatif. Jika Anda memanggil metode Message.nack(), library klien akan menutup koneksi streaming yang ada, dan Anda harus meminta pesan lagi.

Pelanggan harus mengonfirmasi setiap pesan. Pelanggan akan menerima pesan terlama yang belum direspons terlebih dahulu, diikuti dengan setiap pesan berikutnya. Jika pelanggan melewati satu pesan, mengonfirmasi pesan berikutnya, lalu terhubung kembali, pelanggan akan menerima pesan yang tidak dikonfirmasi dan setiap pesan berikutnya yang dikonfirmasi.

Langganan Lite tidak memiliki batas waktu konfirmasi dan layanan Pub/Sub Lite tidak mengirim ulang pesan yang tidak dikonfirmasi melalui koneksi streaming terbuka.

Menggunakan kontrol alur

Setelah layanan Pub/Sub Lite mengirimkan pesan kepada pelanggan, pelanggan akan menyimpan pesan yang tidak terkonfirmasi di memori. Anda dapat membatasi jumlah pesan yang belum selesai yang dapat disimpan pelanggan dalam memori menggunakan setelan kontrol aliran. Setelan kontrol alur berlaku untuk setiap partisi tempat subscriber menerima pesan.

Anda dapat mengonfigurasi setelan kontrol alur berikut:

• Ukuran pesan yang luar biasa. Ukuran maksimum, dalam byte, dari pesan yang belum terkirim. Ukuran maksimum harus lebih besar dari ukuran pesan terbesar.
• Jumlah pesan. Jumlah maksimum pesan yang belum terkirim.

Ukuran pesan ada di kolom size_bytes. Anda dapat mengonfigurasi setelan kontrol alur dengan library klien.

Go

Untuk mengonfigurasi setelan kontrol alur, teruskan ReceiveSettings saat memanggil pscompat.NewSubscriberClientWithSettings. Anda dapat menetapkan parameter berikut di ReceiveSettings:

• MaxOutstandingMessages

• MaxOutstandingBytes

Sebagai contoh, lihat contoh kontrol alur ini.

Java

Untuk mengonfigurasi setelan kontrol alur, gunakan metode berikut di class FlowControlRequest.Builder:

• setBytesOutstanding

• setMessagesOutstanding

Python

Untuk mengonfigurasi setelan kontrol alur, tetapkan parameter berikut di class FlowControlSettings:

• bytes_outstanding

• messages_outstanding

Misalnya, jika jumlah maksimum pesan adalah 100 dan pelanggan terhubung ke 10 partisi, pelanggan tidak dapat menerima lebih dari 100 pesan dari salah satu dari 10 partisi. Jumlah total pesan yang belum terkirim mungkin lebih besar dari 100, tetapi pelanggan tidak dapat menyimpan lebih dari 100 pesan dari setiap partisi.