Nachrichten in Lite-Themen veröffentlichen

Hinweis: Pub/Sub Lite wurde eingestellt. Ab dem 18. März 2026 ist Pub/Sub Lite nicht mehr verfügbar.

• Bestandskunden: Pub/Sub Lite ist noch bis zum 18. März 2026 verfügbar. Wenn Sie Pub/Sub Lite nicht innerhalb der 90 Tage vor dem 24. September 2024 (26. Juni 2024 bis 24. September 2024) verwendet haben, können Sie ab dem 24. September 2024 nicht mehr auf Pub/Sub Lite zugreifen.
• Neukunden: Pub/Sub Lite ist nach dem 24. September 2024 nicht mehr für Neukunden verfügbar.

Sie können Ihren Pub/Sub Lite-Dienst zu Managed Service for Apache Kafka oder Pub/Sub migrieren.

Auf dieser Seite wird erläutert, wie Sie Nachrichten zu Lite-Themen veröffentlichen. Sie können Nachrichten mit der Pub/Sub Lite-Clientbibliothek für Java veröffentlichen.

Nach dem Veröffentlichen von Nachrichten und Erstellen eines Lite-Abos für ein Lite-Thema können Sie Nachrichten vom Lite-Abo empfangen.

Nachrichtenformat

Eine Nachricht besteht aus Feldern mit den Nachrichtendaten und Metadaten. Geben Sie in der Nachricht Folgendes an:

• Nachrichtendaten
• Ein Sortierschlüssel
• Einen Ereigniszeitpunkt
• Attribute mit zusätzlichen Metadaten

Die Clientbibliothek weist die Nachricht automatisch einer Partition zu. Der Pub/Sub Lite-Dienst fügt der Nachricht die folgenden Felder hinzu:

• Eine innerhalb der Partition eindeutige Nachrichten-ID
• Ein Zeitstempel für den Zeitpunkt, zu dem der Pub/Sub Lite-Dienst die Nachricht in der Partition speichert

Nachrichten veröffentlichen

Wenn Sie Nachrichten veröffentlichen möchten, fordern Sie eine Streamingverbindung an das Lite-Thema an und senden Sie dann Nachrichten über die Streamingverbindung.

Hinweis: Das Publisher-Clientobjekt ist langlebig und hält eine Verbindung zu Pub/Sub Lite auf. Dieses Objekt sollte nicht für jede Nachricht erstellt werden. Es sollte für alle Nachrichten zu einem Thema verwendet werden, die aus dem gleichen Programm stammen.

Das folgende Beispiel zeigt, wie Sie Nachrichten in einem Lite-Thema veröffentlichen:

gcloud

Für diesen Befehl ist Python 3.6 oder höher erforderlich und das grpcio-Python-Paket muss installiert sein. Führen Sie für MacOS-, Linux- und Cloud Shell-Nutzer Folgendes aus:

sudo pip3 install grpcio
export CLOUDSDK_PYTHON_SITEPACKAGES=1

Veröffentlichen Sie eine Nachricht mit dem Befehl gcloud pubsub lite-topics publish:

gcloud pubsub lite-topics publish TOPIC_ID \
    --location=LITE_LOCATION \
    --message=MESSAGE_DATA

Dabei gilt:

• TOPIC_ID: die ID des Lite-Themas
• LITE_LOCATION: der Standort des Lite-Themas
• MESSAGE_DATA: ein String mit den Nachrichtendaten

Go

Bevor Sie dieses Beispiel ausführen, folgen Sie der Anleitung zur Einrichtung von Go in Pub/Sub Lite-Clientbibliotheken.

package main

import (
	"context"
	"flag"
	"fmt"
	"log"
	"sync"

	"cloud.google.com/go/pubsub"
	"cloud.google.com/go/pubsublite/pscompat"
	"golang.org/x/sync/errgroup"
)

func main() {
	// NOTE: Set these flags for an existing Pub/Sub Lite topic 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")
	topicID := flag.String("topic_id", "", "Existing Pub/Sub Lite topic")
	messageCount := flag.Int("message_count", 100, "The number of messages to send")
	flag.Parse()

	ctx := context.Background()
	topicPath := fmt.Sprintf("projects/%s/locations/%s/topics/%s", *projectID, *zone, *topicID)

	// Create the publisher client.
	publisher, err := pscompat.NewPublisherClient(ctx, topicPath)
	if err != nil {
		log.Fatalf("pscompat.NewPublisherClient error: %v", err)
	}

	// Ensure the publisher will be shut down.
	defer publisher.Stop()

	// Collect any messages that need to be republished with a new publisher
	// client.
	var toRepublish []*pubsub.Message
	var toRepublishMu sync.Mutex

	// Publish messages. Messages are automatically batched.
	g := new(errgroup.Group)
	for i := 0; i < *messageCount; i++ {
		msg := &pubsub.Message{
			Data: []byte(fmt.Sprintf("message-%d", i)),
		}
		result := publisher.Publish(ctx, msg)

		g.Go(func() error {
			// Get blocks until the result is ready.
			id, err := result.Get(ctx)
			if err != nil {
				// NOTE: A failed PublishResult indicates that the publisher client
				// encountered a fatal error and has permanently terminated. After the
				// fatal error has been resolved, a new publisher client instance must
				// be created to republish failed messages.
				fmt.Printf("Publish error: %v\n", err)
				toRepublishMu.Lock()
				toRepublish = append(toRepublish, msg)
				toRepublishMu.Unlock()
				return err
			}

			// Metadata decoded from the id contains the partition and offset.
			metadata, err := pscompat.ParseMessageMetadata(id)
			if err != nil {
				fmt.Printf("Failed to parse message metadata %q: %v\n", id, err)
				return err
			}
			fmt.Printf("Published: partition=%d, offset=%d\n", metadata.Partition, metadata.Offset)
			return nil
		})
	}
	if err := g.Wait(); err != nil {
		fmt.Printf("Publishing finished with error: %v\n", err)
	}
	fmt.Printf("Published %d messages\n", *messageCount-len(toRepublish))

	// Print the error that caused the publisher client to terminate (if any),
	// which may contain more context than PublishResults.
	if err := publisher.Error(); err != nil {
		fmt.Printf("Publisher client terminated due to error: %v\n", publisher.Error())
	}
}

Java

Bevor Sie dieses Beispiel ausführen, folgen Sie den Schritten zur Einrichtung von Java in Pub/Sub Lite-Clientbibliotheken.

import com.google.api.core.ApiFuture;
import com.google.api.core.ApiFutures;
import com.google.api.gax.rpc.ApiException;
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.TopicName;
import com.google.cloud.pubsublite.TopicPath;
import com.google.cloud.pubsublite.cloudpubsub.Publisher;
import com.google.cloud.pubsublite.cloudpubsub.PublisherSettings;
import com.google.protobuf.ByteString;
import com.google.pubsub.v1.PubsubMessage;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutionException;

public class PublisherExample {

  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 topic for the publish example to work.
    String topicId = "your-topic-id";
    long projectNumber = Long.parseLong("123456789");
    int messageCount = 100;
    // True if using a regional location. False if using a zonal location.
    // https://cloud.google.com/pubsub/lite/docs/topics
    boolean regional = false;

    publisherExample(cloudRegion, zoneId, projectNumber, topicId, messageCount, regional);
  }

  // Publish messages to a topic.
  public static void publisherExample(
      String cloudRegion,
      char zoneId,
      long projectNumber,
      String topicId,
      int messageCount,
      boolean regional)
      throws ApiException, ExecutionException, InterruptedException {

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

    TopicPath topicPath =
        TopicPath.newBuilder()
            .setProject(ProjectNumber.of(projectNumber))
            .setLocation(location)
            .setName(TopicName.of(topicId))
            .build();

    Publisher publisher = null;
    List<ApiFuture<String>> futures = new ArrayList<>();

    try {
      PublisherSettings publisherSettings =
          PublisherSettings.newBuilder().setTopicPath(topicPath).build();

      publisher = Publisher.create(publisherSettings);

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

      for (int i = 0; i < messageCount; i++) {
        String message = "message-" + i;

        // Convert the message to a byte string.
        ByteString data = ByteString.copyFromUtf8(message);
        PubsubMessage pubsubMessage = PubsubMessage.newBuilder().setData(data).build();

        // Publish a message. Messages are automatically batched.
        ApiFuture<String> future = publisher.publish(pubsubMessage);
        futures.add(future);
      }
    } finally {
      ArrayList<MessageMetadata> metadata = new ArrayList<>();
      List<String> ackIds = ApiFutures.allAsList(futures).get();
      for (String id : ackIds) {
        // Decoded metadata contains partition and offset.
        metadata.add(MessageMetadata.decode(id));
      }
      System.out.println(metadata + "\nPublished " + ackIds.size() + " messages.");

      if (publisher != null) {
        // Shut down the publisher.
        publisher.stopAsync().awaitTerminated();
        System.out.println("Publisher is shut down.");
      }
    }
  }
}

Python

Bevor Sie dieses Beispiel ausführen, folgen Sie den Schritten zur Einrichtung von Java in Pub/Sub Lite-Clientbibliotheken.

from google.cloud.pubsublite.cloudpubsub import PublisherClient
from google.cloud.pubsublite.types import (
    CloudRegion,
    CloudZone,
    MessageMetadata,
    TopicPath,
)

# TODO(developer):
# project_number = 1122334455
# cloud_region = "us-central1"
# zone_id = "a"
# topic_id = "your-topic-id"
# regional = True

if regional:
    location = CloudRegion(cloud_region)
else:
    location = CloudZone(CloudRegion(cloud_region), zone_id)

topic_path = TopicPath(project_number, location, topic_id)

# PublisherClient() must be used in a `with` block or have __enter__() called before use.
with PublisherClient() as publisher_client:
    data = "Hello world!"
    api_future = publisher_client.publish(topic_path, data.encode("utf-8"))
    # result() blocks. To resolve API futures asynchronously, use add_done_callback().
    message_id = api_future.result()
    message_metadata = MessageMetadata.decode(message_id)
    print(
        f"Published a message to {topic_path} with partition {message_metadata.partition.value} and offset {message_metadata.cursor.offset}."
    )

Die Clientbibliothek sendet asynchron Nachrichten und verarbeitet Fehler. Wenn ein Fehler auftritt, sendet die Clientbibliothek die Nachricht erneut.

1. Der Pub/Sub Lite-Dienst schließt den Stream.
2. Die Clientbibliothek puffert die Nachrichten und stellt eine Verbindung zum Lite-Thema wieder her.
3. Die Clientbibliothek sendet die Nachrichten der Reihe nach.

Nachdem Sie eine Nachricht veröffentlicht haben, speichert der Pub/Sub Lite-Dienst die Nachricht in einer Partition und gibt die Nachrichten-ID an den Publisher zurück.

Sortierschlüssel verwenden

Wenn Nachrichten denselben Reihenfolgeschlüssel haben, weist die Clientbibliothek die Nachrichten derselben Partition zu. Der Sortierschlüssel darf maximal 1.024 Byte lang sein.

Der Sortierschlüssel befindet sich im Feld key einer Nachricht. Sie können Sortierschlüssel mit der Clientbibliothek festlegen.

gcloud

Für diesen Befehl ist Python 3.6 oder höher erforderlich und das grpcio-Python-Paket muss installiert sein. Führen Sie für MacOS-, Linux- und Cloud Shell-Nutzer Folgendes aus:

sudo pip3 install grpcio
export CLOUDSDK_PYTHON_SITEPACKAGES=1

Veröffentlichen Sie eine Nachricht mit dem Befehl gcloud pubsub lite-topics publish:

gcloud pubsub lite-topics publish TOPIC_ID \
    --location=LITE_LOCATION \
    --ordering-key=ORDERING_KEY \
    --message=MESSAGE_DATA

Dabei gilt:

• TOPIC_ID: die ID des Lite-Themas
• LITE_LOCATION: der Standort des Lite-Themas
• ORDERING_KEY: ein String, mit dem Nachrichten Partitionen zugewiesen werden
• MESSAGE_DATA: ein String mit den Nachrichtendaten

Go

Bevor Sie dieses Beispiel ausführen, folgen Sie der Anleitung zur Einrichtung von Go in Pub/Sub Lite-Clientbibliotheken.

import (
	"context"
	"fmt"
	"io"

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

func publishWithOrderingKey(w io.Writer, projectID, zone, topicID string, messageCount int) error {
	// projectID := "my-project-id"
	// zone := "us-central1-a"
	// topicID := "my-topic"
	// messageCount := 10
	ctx := context.Background()
	topicPath := fmt.Sprintf("projects/%s/locations/%s/topics/%s", projectID, zone, topicID)

	// Create the publisher client.
	publisher, err := pscompat.NewPublisherClient(ctx, topicPath)
	if err != nil {
		return fmt.Errorf("pscompat.NewPublisherClient error: %w", err)
	}

	// Ensure the publisher will be shut down.
	defer publisher.Stop()

	// Messages of the same ordering key will always get published to the same
	// partition. When OrderingKey is unset, messages can get published to
	// different partitions if more than one partition exists for the topic.
	var results []*pubsub.PublishResult
	for i := 0; i < messageCount; i++ {
		r := publisher.Publish(ctx, &pubsub.Message{
			OrderingKey: "test_ordering_key",
			Data:        []byte(fmt.Sprintf("message-%d", i)),
		})
		results = append(results, r)
	}

	// Print publish results.
	var publishedCount int
	for _, r := range results {
		// Get blocks until the result is ready.
		id, err := r.Get(ctx)
		if err != nil {
			// NOTE: A failed PublishResult indicates that the publisher client
			// encountered a fatal error and has permanently terminated. After the
			// fatal error has been resolved, a new publisher client instance must be
			// created to republish failed messages.
			fmt.Fprintf(w, "Publish error: %v\n", err)
			continue
		}

		// Metadata decoded from the id contains the partition and offset.
		metadata, err := pscompat.ParseMessageMetadata(id)
		if err != nil {
			return fmt.Errorf("failed to parse message metadata %q: %w", id, err)
		}
		fmt.Fprintf(w, "Published: partition=%d, offset=%d\n", metadata.Partition, metadata.Offset)
		publishedCount++
	}

	fmt.Fprintf(w, "Published %d messages with ordering key\n", publishedCount)
	return publisher.Error()
}

Java

Bevor Sie dieses Beispiel ausführen, folgen Sie den Schritten zur Einrichtung von Java in Pub/Sub Lite-Clientbibliotheken.

import com.google.api.core.ApiFuture;
import com.google.api.gax.rpc.ApiException;
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.TopicName;
import com.google.cloud.pubsublite.TopicPath;
import com.google.cloud.pubsublite.cloudpubsub.Publisher;
import com.google.cloud.pubsublite.cloudpubsub.PublisherSettings;
import com.google.protobuf.ByteString;
import com.google.pubsub.v1.PubsubMessage;
import java.util.concurrent.ExecutionException;

public class PublishWithOrderingKeyExample {
  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 topic for the publish example to work.
    String topicId = "your-topic-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;

    publishWithOrderingKeyExample(cloudRegion, zoneId, projectNumber, topicId, regional);
  }

  // Publish a message to a topic with an ordering key.
  public static void publishWithOrderingKeyExample(
      String cloudRegion, char zoneId, long projectNumber, String topicId, boolean regional)
      throws ApiException, ExecutionException, InterruptedException {

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

    TopicPath topicPath =
        TopicPath.newBuilder()
            .setProject(ProjectNumber.of(projectNumber))
            .setLocation(location)
            .setName(TopicName.of(topicId))
            .build();

    PublisherSettings publisherSettings =
        PublisherSettings.newBuilder().setTopicPath(topicPath).build();

    Publisher publisher = Publisher.create(publisherSettings);

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

    String message = "message-with-ordering-key";

    // Convert the message to a byte string.
    ByteString data = ByteString.copyFromUtf8(message);
    PubsubMessage pubsubMessage =
        PubsubMessage.newBuilder()
            .setData(data)
            // Messages of the same ordering key will always get published to the
            // same partition. When OrderingKey is unset, messages can get published
            // to different partitions if more than one partition exists for the topic.
            .setOrderingKey("testing")
            .build();

    // Publish a message.
    ApiFuture<String> future = publisher.publish(pubsubMessage);

    // Shut down the publisher.
    publisher.stopAsync().awaitTerminated();

    String ackId = future.get();
    MessageMetadata metadata = MessageMetadata.decode(ackId);
    System.out.println("Published a message with ordering key:\n" + metadata);
  }
}

Python

Bevor Sie dieses Beispiel ausführen, folgen Sie den Schritten zur Einrichtung von Java in Pub/Sub Lite-Clientbibliotheken.

from google.cloud.pubsublite.cloudpubsub import PublisherClient
from google.cloud.pubsublite.types import (
    CloudRegion,
    CloudZone,
    MessageMetadata,
    TopicPath,
)

# TODO(developer):
# project_number = 1122334455
# cloud_region = "us-central1"
# zone_id = "a"
# topic_id = "your-topic-id"
# num_messages = 100
# regional = True

if regional:
    location = CloudRegion(cloud_region)
else:
    location = CloudZone(CloudRegion(cloud_region), zone_id)

topic_path = TopicPath(project_number, location, topic_id)

# PublisherClient() must be used in a `with` block or have __enter__() called before use.
with PublisherClient() as publisher_client:
    for message in range(num_messages):
        data = f"{message}"
        # Messages of the same ordering key will always get published to the same partition.
        # When ordering_key is unset, messsages can get published ot different partitions if
        # more than one partition exists for the topic.
        api_future = publisher_client.publish(
            topic_path, data.encode("utf-8"), ordering_key="testing"
        )
        # result() blocks. To resolve api futures asynchronously, use add_done_callback().
        message_id = api_future.result()
        message_metadata = MessageMetadata.decode(message_id)
        print(
            f"Published {data} to partition {message_metadata.partition.value} and offset {message_metadata.cursor.offset}."
        )

print(
    f"Finished publishing {num_messages} messages with an ordering key to {str(topic_path)}."
)

Mit Reihenfolgeschlüsseln können Sie mehrere Nachrichten an dieselbe Partition senden, sodass Abonnenten die Nachrichten in der Reihenfolge empfangen. Die Clientbibliothek kann einer gleichen Partition mehrere Reihenfolgeschlüssel zuweisen.

Ereigniszeit festlegen

Sie können die Ereigniszeit verwenden, um Ihre Lite-Mitteilungen zu veröffentlichen. „Ereigniszeit“ ist ein benutzerdefiniertes Attribut, das Sie Ihrer Nachricht hinzufügen können.

Sie können den Ereigniszeitstempel mit der Clientbibliothek oder der gCloud CLI festlegen.

Für diesen Befehl ist Python 3.6 oder höher erforderlich und das grpcio-Python-Paket muss installiert sein. Führen Sie für MacOS-, Linux- und Cloud Shell-Nutzer Folgendes aus:

sudo pip3 install grpcio
export CLOUDSDK_PYTHON_SITEPACKAGES=1

Veröffentlichen Sie eine Nachricht mit dem Befehl gcloud pubsub lite-topics publish:

gcloud pubsub lite-topics publish TOPIC_ID \
    --location=LITE_LOCATION \
    --event-time=EVENT_TIME \
    --message=MESSAGE_DATA

Dabei gilt:

• TOPIC_ID: die ID des Lite-Themas

• LITE_LOCATION: der Standort des Lite-Themas

• EVENT_TIME: eine vom Nutzer angegebene Ereigniszeit. Weitere Informationen zu Zeitformaten erhalten Sie durch Eingabe von gcloud topic datetimes.

• MESSAGE_DATA: ein String mit den Nachrichtendaten

Attribute verwenden

Nachrichtenattribute sind Schlüssel/Wert-Paare mit Metadaten zur Nachricht. Die Attribute können Text- oder Byte-Strings sein.

Die Attribute befinden sich im Feld attributes einer Nachricht. Mit der Clientbibliothek können Sie Attribute festlegen.

gcloud

Für diesen Befehl ist Python 3.6 oder höher erforderlich und das grpcio-Python-Paket muss installiert sein. Führen Sie für MacOS-, Linux- und Cloud Shell-Nutzer Folgendes aus:

sudo pip3 install grpcio
export CLOUDSDK_PYTHON_SITEPACKAGES=1

Veröffentlichen Sie eine Nachricht mit dem Befehl gcloud pubsub lite-topics publish:

gcloud pubsub lite-topics publish TOPIC_ID \
    --location=LITE_LOCATION \
    --message=MESSAGE_DATA \
    --attribute=KEY=VALUE,...

Dabei gilt:

• TOPIC_ID: die ID des Lite-Themas
• LITE_LOCATION: der Standort des Lite-Themas
• MESSAGE_DATA: ein String mit den Nachrichtendaten
• KEY ist der Schlüssel eines Nachrichtenattributs
• VALUE: der Wert für den Schlüssel des Nachrichtenattributs

Go

Bevor Sie dieses Beispiel ausführen, folgen Sie der Anleitung zur Einrichtung von Go in Pub/Sub Lite-Clientbibliotheken.

import (
	"context"
	"fmt"
	"io"

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

func publishWithCustomAttributes(w io.Writer, projectID, zone, topicID string) error {
	// projectID := "my-project-id"
	// zone := "us-central1-a"
	// topicID := "my-topic"
	ctx := context.Background()
	topicPath := fmt.Sprintf("projects/%s/locations/%s/topics/%s", projectID, zone, topicID)

	// Create the publisher client.
	publisher, err := pscompat.NewPublisherClient(ctx, topicPath)
	if err != nil {
		return fmt.Errorf("pscompat.NewPublisherClient error: %w", err)
	}

	// Ensure the publisher will be shut down.
	defer publisher.Stop()

	// Publish a message with custom attributes.
	result := publisher.Publish(ctx, &pubsub.Message{
		Data: []byte("message-with-custom-attributes"),
		Attributes: map[string]string{
			"year":   "2020",
			"author": "unknown",
		},
	})

	// Get blocks until the result is ready.
	id, err := result.Get(ctx)
	if err != nil {
		return fmt.Errorf("publish error: %w", err)
	}

	fmt.Fprintf(w, "Published a message with custom attributes: %v\n", id)
	return publisher.Error()
}

Java

Bevor Sie dieses Beispiel ausführen, folgen Sie den Schritten zur Einrichtung von Java in Pub/Sub Lite-Clientbibliotheken.

import com.google.api.core.ApiFuture;
import com.google.api.gax.rpc.ApiException;
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.TopicName;
import com.google.cloud.pubsublite.TopicPath;
import com.google.cloud.pubsublite.cloudpubsub.MessageTransforms;
import com.google.cloud.pubsublite.cloudpubsub.Publisher;
import com.google.cloud.pubsublite.cloudpubsub.PublisherSettings;
import com.google.common.collect.ImmutableMap;
import com.google.protobuf.ByteString;
import com.google.protobuf.util.Timestamps;
import com.google.pubsub.v1.PubsubMessage;
import java.time.Instant;
import java.util.concurrent.ExecutionException;

public class PublishWithCustomAttributesExample {
  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 topic for the publish example to work.
    String topicId = "your-topic-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;

    publishWithCustomAttributesExample(cloudRegion, zoneId, projectNumber, topicId, regional);
  }

  // Publish messages to a topic with custom attributes.
  public static void publishWithCustomAttributesExample(
      String cloudRegion, char zoneId, long projectNumber, String topicId, boolean regional)
      throws ApiException, ExecutionException, InterruptedException {

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

    TopicPath topicPath =
        TopicPath.newBuilder()
            .setProject(ProjectNumber.of(projectNumber))
            .setLocation(location)
            .setName(TopicName.of(topicId))
            .build();

    PublisherSettings publisherSettings =
        PublisherSettings.newBuilder().setTopicPath(topicPath).build();

    Publisher publisher = Publisher.create(publisherSettings);

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

    // Prepare the message data as a byte string.
    String messageData = "message-with-custom-attributes";
    ByteString data = ByteString.copyFromUtf8(messageData);

    // Prepare a protobuf-encoded event timestamp for the message.
    Instant now = Instant.now();
    String eventTime =
        MessageTransforms.encodeAttributeEventTime(Timestamps.fromMillis(now.toEpochMilli()));

    PubsubMessage pubsubMessage =
        PubsubMessage.newBuilder()
            .setData(data)
            // Add two sets of custom attributes to the message.
            .putAllAttributes(ImmutableMap.of("year", "2020", "author", "unknown"))
            // Add an event timestamp as an attribute.
            .putAttributes(MessageTransforms.PUBSUB_LITE_EVENT_TIME_TIMESTAMP_PROTO, eventTime)
            .build();

    // Publish a message.
    ApiFuture<String> future = publisher.publish(pubsubMessage);

    // Shut down the publisher.
    publisher.stopAsync().awaitTerminated();

    String ackId = future.get();
    MessageMetadata metadata = MessageMetadata.decode(ackId);
    System.out.println("Published a message with custom attributes:\n" + metadata);
  }
}

Python

Bevor Sie dieses Beispiel ausführen, folgen Sie den Schritten zur Einrichtung von Java in Pub/Sub Lite-Clientbibliotheken.

from google.cloud.pubsublite.cloudpubsub import PublisherClient
from google.cloud.pubsublite.types import (
    CloudRegion,
    CloudZone,
    MessageMetadata,
    TopicPath,
)

# TODO(developer):
# project_number = 1122334455
# cloud_region = "us-central1"
# zone_id = "a"
# topic_id = "your-topic-id"
# regional = True

if regional:
    location = CloudRegion(cloud_region)
else:
    location = CloudZone(CloudRegion(cloud_region), zone_id)

topic_path = TopicPath(project_number, location, topic_id)

# PublisherClient() must be used in a `with` block or have __enter__() called before use.
with PublisherClient() as publisher_client:
    data = "Hello world!"
    api_future = publisher_client.publish(
        topic_path,
        data.encode("utf-8"),
        year="2020",
        author="unknown",
    )
    # result() blocks. To resolve api futures asynchronously, use add_done_callback().
    message_id = api_future.result()
    message_metadata = MessageMetadata.decode(message_id)
    print(
        f"Published {data} to partition {message_metadata.partition.value} and offset {message_metadata.cursor.offset}."
    )

print(f"Finished publishing a message with custom attributes to {str(topic_path)}.")

Attribute können angeben, wie eine Nachricht verarbeitet werden soll. Abonnenten können das Feld attributes einer Nachricht parsen und die Nachricht entsprechend ihren Attributen verarbeiten.

Nachrichten im Batch verarbeiten

Die Clientbibliothek veröffentlicht Nachrichten im Batch. Größere Batches benötigen weniger Rechenressourcen, erhöhen aber die Latenz. Sie können die Batchgröße mit den Batcheinstellungen ändern.

In der folgenden Tabelle sind die Batch-Einstellungen aufgeführt, die Sie konfigurieren können:

Einstellung	Beschreibung	Default
Größe der Anfrage	Die maximale Größe des Batches in Byte.	3,5 MiB
Zahl der Nachrichten	Die maximale Anzahl von Nachrichten in einem Batch.	1.000 Nachrichten
Verzögerung bei der Veröffentlichung	Die Zeit in Millisekunden zwischen dem Hinzufügen der Nachricht zu einem Batch und dem Senden des Batches an das Lite-Thema.	50 Millisekunden

Sie können Batch-Einstellungen mit der Clientbibliothek konfigurieren.

Go

Bevor Sie dieses Beispiel ausführen, folgen Sie der Anleitung zur Einrichtung von Go in Pub/Sub Lite-Clientbibliotheken.

import (
	"context"
	"fmt"
	"io"
	"time"

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

func publishWithBatchSettings(w io.Writer, projectID, zone, topicID string, messageCount int) error {
	// projectID := "my-project-id"
	// zone := "us-central1-a"
	// topicID := "my-topic"
	// messageCount := 10
	ctx := context.Background()
	topicPath := fmt.Sprintf("projects/%s/locations/%s/topics/%s", projectID, zone, topicID)

	// Batch settings control how the publisher batches messages. These settings
	// apply per partition.
	// See https://pkg.go.dev/cloud.google.com/go/pubsublite/pscompat#pkg-variables
	// for DefaultPublishSettings.
	settings := pscompat.PublishSettings{
		ByteThreshold:  5 * 1024, // 5 KiB
		CountThreshold: 1000,     // 1,000 messages
		DelayThreshold: 100 * time.Millisecond,
	}

	// Create the publisher client.
	publisher, err := pscompat.NewPublisherClientWithSettings(ctx, topicPath, settings)
	if err != nil {
		return fmt.Errorf("pscompat.NewPublisherClientWithSettings error: %w", err)
	}

	// Ensure the publisher will be shut down.
	defer publisher.Stop()

	// Publish requests are sent to the server based on request size, message
	// count and time since last publish, whichever condition is met first.
	var results []*pubsub.PublishResult
	for i := 0; i < messageCount; i++ {
		r := publisher.Publish(ctx, &pubsub.Message{
			Data: []byte(fmt.Sprintf("message-%d", i)),
		})
		results = append(results, r)
	}

	// Print publish results.
	var publishedCount int
	for _, r := range results {
		// Get blocks until the result is ready.
		id, err := r.Get(ctx)
		if err != nil {
			// NOTE: A failed PublishResult indicates that the publisher client
			// encountered a fatal error and has permanently terminated. After the
			// fatal error has been resolved, a new publisher client instance must be
			// created to republish failed messages.
			fmt.Fprintf(w, "Publish error: %v\n", err)
			continue
		}
		fmt.Fprintf(w, "Published: %v\n", id)
		publishedCount++
	}

	fmt.Fprintf(w, "Published %d messages with batch settings\n", publishedCount)
	return publisher.Error()
}

Java

Bevor Sie dieses Beispiel ausführen, folgen Sie den Schritten zur Einrichtung von Java in Pub/Sub Lite-Clientbibliotheken.

import com.google.api.core.ApiFuture;
import com.google.api.core.ApiFutures;
import com.google.api.gax.batching.BatchingSettings;
import com.google.api.gax.rpc.ApiException;
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.TopicName;
import com.google.cloud.pubsublite.TopicPath;
import com.google.cloud.pubsublite.cloudpubsub.Publisher;
import com.google.cloud.pubsublite.cloudpubsub.PublisherSettings;
import com.google.protobuf.ByteString;
import com.google.pubsub.v1.PubsubMessage;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutionException;
import org.threeten.bp.Duration;

public class PublishWithBatchSettingsExample {
  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 topic for the publish example to work.
    String topicId = "your-topic-id";
    long projectNumber = Long.parseLong("123456789");
    int messageCount = 100;
    // True if using a regional location. False if using a zonal location.
    // https://cloud.google.com/pubsub/lite/docs/topics
    boolean regional = false;

    publishWithBatchSettingsExample(
        cloudRegion, zoneId, projectNumber, topicId, messageCount, regional);
  }

  // Publish messages to a topic with batch settings.
  public static void publishWithBatchSettingsExample(
      String cloudRegion,
      char zoneId,
      long projectNumber,
      String topicId,
      int messageCount,
      boolean regional)
      throws ApiException, ExecutionException, InterruptedException {

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

    TopicPath topicPath =
        TopicPath.newBuilder()
            .setProject(ProjectNumber.of(projectNumber))
            .setLocation(location)
            .setName(TopicName.of(topicId))
            .build();

    Publisher publisher = null;
    List<ApiFuture<String>> futures = new ArrayList<>();

    try {
      // Batch settings control how the publisher batches messages
      long requestBytesThreshold = 5000L; // default : 3_500_000 bytes
      long messageCountBatchSize = 100L; // default : 1000L message
      Duration publishDelayThreshold = Duration.ofMillis(100); // default : 50 ms

      // Publish request get triggered based on request size, messages count & time since last
      // publish, whichever condition is met first.
      BatchingSettings batchingSettings =
          BatchingSettings.newBuilder()
              .setRequestByteThreshold(requestBytesThreshold)
              .setElementCountThreshold(messageCountBatchSize)
              .setDelayThreshold(publishDelayThreshold)
              .build();

      PublisherSettings publisherSettings =
          PublisherSettings.newBuilder()
              .setTopicPath(topicPath)
              .setBatchingSettings(batchingSettings)
              .build();

      publisher = Publisher.create(publisherSettings);

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

      for (int i = 0; i < messageCount; i++) {
        String message = "message-" + i;

        // Convert the message to a byte string.
        ByteString data = ByteString.copyFromUtf8(message);
        PubsubMessage pubsubMessage = PubsubMessage.newBuilder().setData(data).build();

        // Publish a message.
        ApiFuture<String> future = publisher.publish(pubsubMessage);
        futures.add(future);
      }
    } finally {
      ArrayList<MessageMetadata> metadata = new ArrayList<>();
      List<String> ackIds = ApiFutures.allAsList(futures).get();
      System.out.println("Published " + ackIds.size() + " messages with batch settings.");

      if (publisher != null) {
        // Shut down the publisher.
        publisher.stopAsync().awaitTerminated();
      }
    }
  }
}

Python

Bevor Sie dieses Beispiel ausführen, folgen Sie den Schritten zur Einrichtung von Java in Pub/Sub Lite-Clientbibliotheken.

from google.cloud.pubsub_v1.types import BatchSettings
from google.cloud.pubsublite.cloudpubsub import PublisherClient
from google.cloud.pubsublite.types import (
    CloudRegion,
    CloudZone,
    MessageMetadata,
    TopicPath,
)

# TODO(developer):
# project_number = 1122334455
# cloud_region = "us-central1"
# zone_id = "a"
# topic_id = "your-topic-id"
# num_messages = 100
# regional = True

if regional:
    location = CloudRegion(cloud_region)
else:
    location = CloudZone(CloudRegion(cloud_region), zone_id)

topic_path = TopicPath(project_number, location, topic_id)
batch_setttings = BatchSettings(
    # 2 MiB. Default to 3 MiB. Must be less than 4 MiB gRPC's per-message limit.
    max_bytes=2 * 1024 * 1024,
    # 100 ms. Default to 50 ms.
    max_latency=0.1,
    # Default to 1000.
    max_messages=100,
)

# PublisherClient() must be used in a `with` block or have __enter__() called before use.
with PublisherClient(
    per_partition_batching_settings=batch_setttings
) as publisher_client:
    for message in range(num_messages):
        data = f"{message}"
        api_future = publisher_client.publish(topic_path, data.encode("utf-8"))
        # result() blocks. To resolve API futures asynchronously, use add_done_callback().
        message_id = api_future.result()
        message_metadata = MessageMetadata.decode(message_id)
        print(
            f"Published {data} to partition {message_metadata.partition.value} and offset {message_metadata.cursor.offset}."
        )

print(
    f"Finished publishing {num_messages} messages with batch settings to {str(topic_path)}."
)

Wenn eine Publisher-Anwendung gestartet wird, erstellt die Client-Bibliothek einen Batch für jede Partition in einem Lite-Thema. Wenn ein Lite-Thema beispielsweise zwei Partitionen hat, erstellen Publisher zwei Batches und senden jeden Batch an eine Partition.

Nachdem Sie eine Nachricht veröffentlicht haben, wird sie von der Clientbibliothek gepuffert, bis der Batch die maximale Anforderungsgröße, die maximale Anzahl von Nachrichten oder die Veröffentlichungsverzögerung überschreitet.

Nachrichten in eine Reihenfolge bringen

Lite-Themen sortieren Nachrichten in jeder Partition, indem Sie die Nachrichten veröffentlichen. Verwenden Sie einen Reihenfolgeschlüssel, um Nachrichten derselben Partition zuzuweisen.

Pub/Sub Lite liefert die Nachrichten aus einer Partition nacheinander und Abonnenten können die Nachrichten der Reihe nach verarbeiten. Weitere Informationen finden Sie unter Nachrichten empfangen.

Idempotenz der Veröffentlichung

Pub/Sub Lite-Clientbibliotheken unterstützen das idempotente Veröffentlichen ab den folgenden Versionen:

• java-pubsublite: Version 1.10.0.
• python-pubsublite: Version 1.8.0.
• google-cloud-go: pubsublite Version 1.7.0.

Wenn die Veröffentlichung einer Nachricht aufgrund von Netzwerk- oder Serverfehlern wiederholt wird, wird sie genau einmal gespeichert. Die Idempotency ist nur innerhalb derselben Sitzung gewährleistet. Sie kann nicht garantiert werden, wenn dieselbe Nachricht mit einem neuen Publisher-Client noch einmal veröffentlicht wird. Es fallen keine zusätzlichen Servicekosten an und die Veröffentlichungslatenz wird nicht erhöht.

Idempotentes Veröffentlichen aktivieren oder deaktivieren

Das idempotente Publizieren ist in den Pub/Sub Lite-Clientbibliotheken standardmäßig aktiviert. Sie kann mit den Publisher-Clienteinstellungen in der jeweiligen Clientbibliothek deaktiviert werden.

Wenn die idempotente Veröffentlichung aktiviert ist, kann der in einem Veröffentlichungsergebnis zurückgegebene Offset -1 sein. Dieser Wert wird zurückgegeben, wenn die Nachricht als Duplikat einer bereits erfolgreich veröffentlichten Nachricht erkannt wird, der Server aber nicht genügend Informationen hat, um den Offset der Nachricht zum Zeitpunkt der Veröffentlichung zurückzugeben. Nachrichten, die von Abonnenten empfangen werden, haben immer einen gültigen Offset.

Fehlerbehebung

Erhaltene Duplikate

Da die Idempotenz auf eine einzelne Sitzung beschränkt ist, werden möglicherweise Duplikate empfangen, wenn Sie den Publisher-Client neu erstellen, um dieselben Nachrichten zu veröffentlichen.

Ein Abonnentenclient kann dieselbe Nachricht mehrmals erhalten, wenn Abonnenten vom Pub/Sub Lite-Dienst automatisch Partitionen zugewiesen werden (Standardeinstellung). Eine Nachricht wird möglicherweise an einen anderen Abonnentenclient weitergeleitet, wenn eine Neuzuweisung erfolgt.

Fehler des Publishers

Der Status einer Publisher-Sitzung wird nach 7 Tagen Inaktivität auf dem Server automatisch beseitigt. Wenn eine Sitzung nach diesem Zeitraum fortgesetzt wird, wird der Publisher-Client mit einer Fehlermeldung wie „Vorbedingung fehlgeschlagen: Die Nachricht sollte die Veröffentlichungssequenznummer … haben“ beendet und akzeptiert keine neuen Nachrichten. Erstelle den Publisher-Client neu, um diesen Fehler zu beheben.