创建和管理精简版主题

本文档介绍如何创建、查看和删除 Pub/Sub Lite 主题。

精简版主题概览

Pub/Sub Lite 服务需要一个主题，发布者可以向该主题发送消息，以便订阅者可以接收这些消息。在 Pub/Sub Lite 中，您可以选择创建可用区级或区域级 Lite 主题。

可用区级精简版和区域级精简版主题仅在复制数据的方式上有所不同。所有其他主题属性与本文档后面部分中介绍的相同。

精简版主题中的数据复制

Pub/Sub Lite 区域级主题将数据存储在单个区域的两个可用区中。Pub/Sub 精简版可用区级主题仅在一个可用区内复制数据。Pub/Sub Lite 异步复制数据。

Pub/Sub Lite 的可用性取决于其底层组件（包括应用前端和存储）的可用性。

Pub/Sub 精简版可用区级主题的可用性取决于所配置可用区中应用前端和存储空间的可用性。如果应用前端或存储发生可用区级故障，可用区级精简版主题在服务中断期间将不可用。

Pub/Sub Lite 区域级主题的可用性取决于应用前端和存储空间在其配置的区域的可用性。可能会出现以下类型的失败：

  • 如果应用前端或存储遇到可用区级故障，则区域级精简版主题仍然可用。

  • 如果应用前端和存储服务都遇到可用区级故障，则区域级精简版主题将仍然可用，前提是这两个组件不会同时开始发生故障。

  • 如果应用前端和存储空间同时发生故障，则区域级精简版主题在服务中断期间将仍然不可用。在此期间，客户端可能无法发布消息或使用消息，但消息仍会按正确的顺序传送。

区域级精简版主题可以防范大多数类型的单可用区故障。但在极少数情况下，复制的异步性质可能会导致单个可用区中的多个组件同时发生故障。因此，区域级精简版主题将不可用。

精简版主题的属性

精简版主题具有以下属性：

  • 分区数。分区是 Pub/Sub Lite 中并行处理的基本单元。一个精简版主题可以有一个或多个分区。

  • 分区的容量。分区的容量由以下三个属性描述：

    • 发布吞吐量 (MiBps)。可以发布消息的最大速率。

    • 订阅吞吐量 (MiBps)。向 Lite 订阅转发消息的最大速率。

    • 存储空间 (GiB)。分区中消息的大小上限。 分区的可配置最小容量为 30 GiB。

  • 精简版主题的存储空间容量。精简版主题中的所有分区都具有相同的已配置存储容量。精简版主题的总存储容量是该主题中所有分区的存储容量之和。例如，如果精简版主题有 8 个分区，每个分区的大小为 30 GiB，则精简版主题的总存储容量为 240 GiB (8 x 30)。

  • 精简版主题的吞吐量容量。吞吐量容量由精简版主题中所有分区的总发布和订阅吞吐量容量组成。即使某个主题总体可以支持 1 GiBps 的发布吞吐量，也是每个分区的发布限制实际上会限制吞吐量。

  • 订阅者客户端和订阅的限制。Pub/Sub Lite 支持为单个分区和单个订阅使用单个订阅者客户端。例如，一个订阅包含 5 个分区的主题最多可以有 5 个订阅者客户端，这些订阅者客户端使用来自该主题的消息。如果配置了更多订阅者客户端，多余的客户端将保持空闲状态。包含 2 个订阅的同一主题最多可以有 10 个订阅者客户端，每个订阅最多可以有 5 个客户端。

  • 预留。预留是为多个主题预配和管理吞吐量容量的最简单方法。对于区域级精简版，预配吞吐量容量是必需的；对于可用区级精简版主题，这是可选项。

    如果您没有为可用区级精简版主题使用预留，则必须明确预配主题的发布和订阅吞吐量容量。如果不使用预留，精简版主题中的所有分区都将具有相同的已配置吞吐量容量。如需了解详情，请参阅在没有预留的情况下配置吞吐量容量。

  • 存储配置。如需配置存储，您必须同时指定存储容量和（可选）消息保留期限。

    • 消息保留期限。Lite 主题存储消息的最长时间。如果您未指定消息保留期限，则精简版主题将存储消息，直至您超出存储容量。

    • 存储空间容量。Pub/Sub Lite 主题容量由主题中的分区数量和每个分区的存储空间容量决定。每个分区的存储空间是指每个分区的存储空间量（以字节为单位）。请注意，区域精简版主题会为发布的每个字节使用两个字节的存储空间。可用区级精简版主题针对发布的每个字节使用 1 字节的存储容量。

    如果您超出存储容量，Pub/Sub Lite 服务会继续发布消息。不过，为了增加存储空间容量，系统会删除分区中最早的消息。即使消息在消息保留期限内，也会删除最早的消息。该服务绝不会在消息的 TTL 到期之前删除消息，除非必须移除这些消息以便为新发布的消息腾出空间。如需了解容量与价格的关系，请参阅 Pub/Sub 精简版价格。

主题命名准则

Pub/Sub Lite 资源名称唯一标识 Pub/Sub Lite 资源，例如主题、订阅或预留。资源名称必须符合以下格式：

projects/project-identifier/collection/ID

  • project-identifier：必须是可从 Google Cloud 控制台获得的项目 ID 或项目编号。例如，my-cool-project 是项目 ID。 123456789123 是项目编号。

  • collection：必须是 topics、subscriptions 或 reservations 中的一个。

  • ID：必须符合以下准则：

    • 不以字符串 goog 开头
    • 以字母开头
    • 包含 3 到 255 个字符
    • 只能包含以下字符：字母 [A-Za-z]、数字 [0-9]、短划线 -、下划线 _、句点 .、波浪线 ~、加号 + 和百分号 %

    您可以在不进行网址编码的资源名称中使用上述列表中的特殊字符。不过，在网址中使用任何其他特殊字符时，您必须确保已正确编码或解码这些字符。例如，mi-tópico 就是无效的 ID。不过，mi-t%C3%B3pico 是有效的。当您进行 REST 调用时，此格式很重要。

提高精简版主题的吞吐量容量

如果您未使用精简版预留，则可以将任何可用区级精简版主题的每个分区的吞吐量容量设置为以下限制所设置的值：

  • 发布吞吐量介于 4 到 16 MiBps 之间

  • 订阅吞吐量介于 4 到 32 MiBps 之间

这些限制也适用于与精简版预留关联的分区。使用预留时，只要不超过预留的总容量，每个分区都可以在指定限制范围内使用任何吞吐量容量。（可选）您可以为每个分区的实际吞吐量设置下限，以提高流量的可预测性。

您可以增加某个主题的分区数量，对其进行横向扩缩。主题中的分区数量无法减少。

更改分区数时，系统不会保留消息的相对顺序。在内部，消息使用哈希函数分配到分区。更新主题中的分区数会更改此哈希函数。在分区数更新后发布具有相同键的消息可能会映射到与更新之前发布的分区不同的分区。

此外，主题调整大小需要几秒钟才能传播到所有发布商。在一段时间内，有些发布商会使用新的哈希函数，而有些发布商仍在使用旧函数。

增加精简版主题的存储容量

您可以增加或减少精简版主题中的存储空间量。精简版主题为每个分区预配相同数量的存储。如果您将存储空间增加到 60 GiB，则每个分区都会获得 60 GiB 的存储空间。

如果您减少精简版主题的存储量，Pub/Sub Lite 服务会首先删除最早的消息。

如需了解容量与价格的关系，请参阅 Pub/Sub 精简版价格

创建精简版主题

您可以使用 Google Cloud 控制台、Google Cloud CLI 或 Pub/Sub Lite API 创建精简版主题。

如需查看可用区域和可用区的列表，请参阅 Pub/Sub Lite 位置。

控制台

  1. 前往精简版主题页面。

    前往精简版主题

  2. 点击创建精简版主题。

    1. 如需创建可用区级精简版主题，请选择一个区域和一个可用区内的可用区。

    2. 如需创建区域级精简版主题，请选择一个区域。

    主题创建后，您便无法更新地理位置。

  3. 在名称部分，输入精简版主题 ID。

    精简版主题名称包含精简版主题 ID、位置和项目编号。

  4. 在吞吐量部分中，执行以下操作：

    1. 选择或创建预留。

      对于可用区级精简版主题，这是可选项；对于区域级精简版主题，这是必填项。

    2. 输入发布吞吐量峰值和订阅吞吐量峰值。

      许多分区都是根据您的输入自动设置的。保存主题后，您可以修改此数量。

  5. 在 Message storage 部分：

    1. 输入每个分区的存储空间。

    2. 选择消息保留期限的两个选项之一。

  6. 检查相应更改，然后点击创建。

gcloud

如需创建精简版主题，请使用 gcloud pubsub lite-topics create 命令：

gcloud pubsub lite-topics create TOPIC_ID \
  --location=LOCATION \
  --partitions=NUMBER_OF_PARTITIONS \
  --per-partition-bytes=STORAGE_PER_PARTITION \
  [--throughput-reservation=RESERVATION_NAME \]
  [--message-retention-period=MESSAGE_RETENTION_PERIOD]

替换以下内容：

  • TOPIC_ID：精简版主题的 ID

  • LOCATION：Pub/Sub 精简版支持的位置的名称

  • NUMBER_OF_PARTITIONS：精简主题中分区数的整数

  • STORAGE_PER_PARTITION：每个分区的存储空间量，例如 30GiB

  • RESERVATION_NAME：精简版预留的完全限定路径，例如 projects/123456789/locations/us-west1/reservations/my-reservation

  • MESSAGE_RETENTION_PERIOD：精简版主题存储消息的时间，例如 1d 或 2w

如果请求成功，命令行会显示一条确认消息：

Created [TOPIC_ID].

协议

要创建精简版主题，请发送 POST 请求，如下所示：

POST https://REGION-pubsublite.googleapis.com/v1/admin/projects/PROJECT_NUMBER/locations/LOCATION/topics/TOPIC_ID
Authorization: Bearer $(gcloud auth print-access-token)

替换以下内容：

在请求正文中指定以下字段：

{
  "partitionConfig": {
       "count": NUMBER_OF_PARTITIONS,
       "capacity": {
         "publishMibPerSec": publish_CAPACITY,
         "subscribeMibPerSec": subscribe_CAPACITY,
       }
  },
  "retentionConfig": {
       "perPartitionBytes": STORAGE_PER_PARTITION,
       "period": MESSAGE_RETENTION_PERIOD,
  },
  "reservationConfig": {
       "throughputReservation": RESERVATION_NAME
  }
}

替换以下内容：

  • NUMBER_OF_PARTITIONS：精简主题中分区数的整数

  • STORAGE_PER_PARTITION：每个分区的存储空间量，例如 30GiB

  • publish_CAPACITY：一个整数，表示每个分区的发布吞吐量容量

  • subscribe_CAPACITY：一个整数，表示每个分区的订阅吞吐量容量

  • MESSAGE_RETENTION_PERIOD：精简版主题存储消息的时间，例如 1d 或 2w

  • RESERVATION_NAME：精简版预留的完全限定路径，例如 projects/123456789/locations/us-west1/reservations/my-reservation

如果请求成功，则响应是 JSON 格式的精简版主题：

{
  "name": projects/PROJECT_NUMBER/locations/LOCATION/topics/TOPIC_ID,
  "partitionConfig": {
       "count": NUMBER_OF_PARTITIONS,
       "capacity": {
         "publishMibPerSec": publish_CAPACITY,
         "subscribeMibPerSec": subscribe_CAPACITY,
       }
  },
  "retentionConfig": {
       "perPartitionBytes": STORAGE_PER_PARTITION,
       "period": MESSAGE_RETENTION_PERIOD,
  },
  "reservationConfig": {
       "throughputReservation": RESERVATION_NAME
  }
}

Go

运行此示例之前，请按照 Pub/Sub 精简版客户端库中的 Go 设置说明进行操作。

import (
	"context"
	"fmt"
	"io"

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

func createTopic(w io.Writer, projectID, region, location, topicID, reservation string) error {
	// projectID := "my-project-id"
	// region := "us-central1" // see https://cloud.google.com/pubsub/lite/docs/locations
	// NOTE: location can be either a region ("us-central1") or a zone ("us-central1-a")
	// For a list of valid locations, see https://cloud.google.com/pubsub/lite/docs/locations.
	// location := "us-central1"
	// topicID := "my-topic"
	// reservation := "projects/my-project-id/reservations/my-reservation"
	ctx := context.Background()
	client, err := pubsublite.NewAdminClient(ctx, region)
	if err != nil {
		return fmt.Errorf("pubsublite.NewAdminClient: %w", err)
	}
	defer client.Close()

	const gib = 1 << 30

	topicPath := fmt.Sprintf("projects/%s/locations/%s/topics/%s", projectID, location, topicID)
	// For ranges of fields in TopicConfig, see https://pkg.go.dev/cloud.google.com/go/pubsublite/#TopicConfig
	topic, err := client.CreateTopic(ctx, pubsublite.TopicConfig{
		Name:                       topicPath,
		PartitionCount:             2, // Must be >= 1 and cannot decrease after creation.
		PublishCapacityMiBPerSec:   4,
		SubscribeCapacityMiBPerSec: 8,
		PerPartitionBytes:          30 * gib,
		RetentionDuration:          pubsublite.InfiniteRetention,
		ThroughputReservation:      reservation,
	})
	if err != nil {
		return fmt.Errorf("client.CreateTopic got err: %w", err)
	}
	fmt.Fprintf(w, "Created topic: %#v\n", topic)
	return nil
}

Java

在运行此示例之前，请按照 Pub/Sub Lite 客户端库中的 Java 设置说明进行操作。

import com.google.api.gax.rpc.AlreadyExistsException;
import com.google.cloud.pubsublite.AdminClient;
import com.google.cloud.pubsublite.AdminClientSettings;
import com.google.cloud.pubsublite.CloudRegion;
import com.google.cloud.pubsublite.CloudRegionOrZone;
import com.google.cloud.pubsublite.CloudZone;
import com.google.cloud.pubsublite.ProjectNumber;
import com.google.cloud.pubsublite.ReservationName;
import com.google.cloud.pubsublite.ReservationPath;
import com.google.cloud.pubsublite.TopicName;
import com.google.cloud.pubsublite.TopicPath;
import com.google.cloud.pubsublite.proto.Topic;
import com.google.cloud.pubsublite.proto.Topic.PartitionConfig;
import com.google.cloud.pubsublite.proto.Topic.PartitionConfig.Capacity;
import com.google.cloud.pubsublite.proto.Topic.ReservationConfig;
import com.google.cloud.pubsublite.proto.Topic.RetentionConfig;
import com.google.protobuf.util.Durations;
import java.util.concurrent.ExecutionException;

public class CreateTopicExample {

  public static void main(String... args) throws Exception {
    // TODO(developer): Replace these variables before running the sample.
    String cloudRegion = "your-cloud-region";
    char zoneId = 'a';
    String topicId = "your-topic-id";
    String reservationId = "your-reservation-id";
    long projectNumber = Long.parseLong("123456789");
    int partitions = 1;
    // True if using a regional location. False if using a zonal location.
    // https://cloud.google.com/pubsub/lite/docs/topics
    boolean regional = false;

    createTopicExample(
        cloudRegion, zoneId, projectNumber, topicId, reservationId, partitions, regional);
  }

  public static void createTopicExample(
      String cloudRegion,
      char zoneId,
      long projectNumber,
      String topicId,
      String reservationId,
      int partitions,
      boolean regional)
      throws Exception {

    ReservationPath reservationPath =
        ReservationPath.newBuilder()
            .setProject(ProjectNumber.of(projectNumber))
            .setLocation(CloudRegion.of(cloudRegion))
            .setName(ReservationName.of(reservationId))
            .build();

    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();

    Topic topic =
        Topic.newBuilder()
            .setPartitionConfig(
                PartitionConfig.newBuilder()
                    // Set throughput capacity per partition in MiB/s.
                    .setCapacity(
                        Capacity.newBuilder()
                            // Must be 4-16 MiB/s.
                            .setPublishMibPerSec(4)
                            // Must be 4-32 MiB/s.
                            .setSubscribeMibPerSec(8)
                            .build())
                    .setCount(partitions))
            .setRetentionConfig(
                RetentionConfig.newBuilder()
                    // How long messages are retained.
                    .setPeriod(Durations.fromDays(1))
                    // Set storage per partition to 30 GiB. This must be 30 GiB-10 TiB.
                    // If the number of bytes stored in any of the topic's partitions grows
                    // beyond this value, older messages will be dropped to make room for
                    // newer ones, regardless of the value of `period`.
                    .setPerPartitionBytes(30 * 1024 * 1024 * 1024L))
            .setReservationConfig(
                ReservationConfig.newBuilder()
                    .setThroughputReservation(reservationPath.toString())
                    .build())
            .setName(topicPath.toString())
            .build();

    AdminClientSettings adminClientSettings =
        AdminClientSettings.newBuilder().setRegion(CloudRegion.of(cloudRegion)).build();

    try (AdminClient adminClient = AdminClient.create(adminClientSettings)) {
      Topic response = adminClient.createTopic(topic).get();
      if (regional) {
        System.out.println(response.getAllFields() + " (regional topic) created successfully.");
      } else {
        System.out.println(response.getAllFields() + " (zonal topic) created successfully.");
      }
    } catch (ExecutionException e) {
      try {
        throw e.getCause();
      } catch (AlreadyExistsException alreadyExists) {
        System.out.println("This topic already exists.");
      } catch (Throwable throwable) {
        throwable.printStackTrace();
      }
    }
  }
}

Python

在运行此示例之前，请按照 Pub/Sub Lite 客户端库中的 Python 设置说明进行操作。

from google.api_core.exceptions import AlreadyExists
from google.cloud.pubsublite import AdminClient, Topic
from google.cloud.pubsublite.types import (
    CloudRegion,
    CloudZone,
    ReservationPath,
    TopicPath,
)
from google.protobuf.duration_pb2 import Duration

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

cloud_region = CloudRegion(cloud_region)
reservation_path = ReservationPath(project_number, cloud_region, reservation_id)

topic_path = None
if regional:
    #  A regional topic.
    topic_path = TopicPath(project_number, cloud_region, topic_id)
else:
    #  A zonal topic
    topic_path = TopicPath(
        project_number, CloudZone(cloud_region, zone_id), topic_id
    )

topic = Topic(
    name=str(topic_path),
    partition_config=Topic.PartitionConfig(
        # A topic must have at least one partition.
        count=num_partitions,
        # Set throughput capacity per partition in MiB/s.
        capacity=Topic.PartitionConfig.Capacity(
            # Set publish throughput capacity per partition to 4 MiB/s. Must be >= 4 and <= 16.
            publish_mib_per_sec=4,
            # Set subscribe throughput capacity per partition to 4 MiB/s. Must be >= 4 and <= 32.
            subscribe_mib_per_sec=8,
        ),
    ),
    retention_config=Topic.RetentionConfig(
        # Set storage per partition to 30 GiB. This must be in the range 30 GiB-10TiB.
        # If the number of byptes stored in any of the topic's partitions grows beyond
        # this value, older messages will be dropped to make room for newer ones,
        # regardless of the value of `period`.
        per_partition_bytes=30 * 1024 * 1024 * 1024,
        # Allow messages to be retained for 7 days.
        period=Duration(seconds=60 * 60 * 24 * 7),
    ),
    reservation_config=Topic.ReservationConfig(
        throughput_reservation=str(reservation_path),
    ),
)

client = AdminClient(cloud_region)
try:
    response = client.create_topic(topic)
    if regional:
        print(f"{response.name} (regional topic) created successfully.")
    else:
        print(f"{response.name} (zonal topic) created successfully.")
except AlreadyExists:
    print(f"{topic_path} already exists.")

创建精简版主题后，您可以扩缩每个分区的吞吐量和存储容量。您也可以增加（但不能减少）分区数。

更新精简版主题

您可以更新精简版主题的以下属性：

  • 与精简版主题关联的预留

  • 主题中的分区数量（只能增加）

  • 发布和订阅吞吐量容量限制

  • 每个分区的存储空间

  • 消息保留类型

在扩缩主题的吞吐量容量或存储容量之前，请参阅扩缩吞吐量容量和扩缩存储容量。

您可以使用 Google Cloud 控制台、Google Cloud CLI 或 Pub/Sub Lite API 更新精简版主题。

控制台

  1. 前往精简版主题页面。

    前往精简版主题

  2. 点击精简版主题 ID。

  3. 点击修改。

  4. 您可以执行以下任何或所有更新：

    • 使用其他预留。对于可用区级主题，您可以选择不使用预留。
    • 输入分区数量。此值可以增加但不能减少。
    • 启用或停用允许的发布和订阅吞吐量上限。
    • 对于可用区级主题，请更新预留的发布和订阅吞吐量。对于附加了预留的主题，更新后的吞吐量会反映新的上限。
    • 修改每个分区的存储空间容量。
    • 修改消息保留期限。

  5. 点击更新。

gcloud

要更新精简版主题，请使用 gcloud pubsub lite-topics update 命令：

gcloud pubsub lite-topics update TOPIC_ID \
  --location=LOCATION \
  --partitions=NUMBER_OF_PARTITIONS \
  --per-partition-publish-mib=publish_CAPACITY \
  --per-partition-subscribe-mib=subscribe_CAPACITY \
  --per-partition-bytes=STORAGE_PER_PARTITION \
  --message-retention-period=MESSAGE_RETENTION_PERIOD

替换以下内容：

  • TOPIC_ID：精简版主题的 ID

  • LOCATION：精简版主题所在的位置名称

  • NUMBER_OF_PARTITIONS：要为精简版主题配置的分区数。

  • publish_CAPACITY：一个整数，表示每个分区的发布吞吐量容量

  • subscribe_CAPACITY：一个整数，表示每个分区的订阅吞吐量容量

  • STORAGE_PER_PARTITION：每个分区的存储空间量，例如 30GiB

  • MESSAGE_RETENTION_PERIOD：精简版主题存储消息的时间，例如 1d 或 2w

如果请求成功，命令行将显示精简版主题：

name: projects/PROJECT_NUMBER/locations/LOCATION/topics/TOPIC_ID
partitionConfig:
  count: NUMBER_OF_PARTITIONS
  capacity:
    publishMibPerSec: publish_CAPACITY
    subscribeMibPerSec: subscribe_CAPACITY
retentionConfig:
  perPartitionBytes: STORAGE_PER_PARTITION
  period: MESSAGE_RETENTION_PERIOD

协议

要更新精简版主题，请发送 PATCH 请求，如下所示：

PATCH https://REGION-pubsublite.googleapis.com/v1/admin/projects/PROJECT_NUMBER/locations/LOCATION/topics/TOPIC_ID?updateMask=partitionConfig.capacity,retentionConfig.perPartitionBytes,retentionConfig.period,partitionConfig.count
Authorization: Bearer $(gcloud auth print-access-token)

替换以下内容：

  • REGION：精简版主题所在的区域

  • PROJECT_NUMBER：具有精简版主题的项目的项目编号

  • LOCATION：精简版主题所在的位置名称

  • TOPIC_ID：精简版主题的 ID

在请求正文中指定以下字段：

{
  "partitionConfig": {
      "count": NUMBER_OF_PARTITIONS,
      "capacity": {
         "publishMibPerSec": publish_CAPACITY,
         "subscribeMibPerSec": subscribe_CAPACITY,
      }
   },
   "retentionConfig": {
       "perPartitionBytes": STORAGE_PER_PARTITION,
       "period": MESSAGE_RETENTION_PERIOD,
   },
}

替换以下内容：

  • publish_CAPACITY：一个整数，表示每个分区的发布吞吐量容量

  • subscribe_CAPACITY：一个整数，表示每个分区的订阅吞吐量容量

  • STORAGE_PER_PARTITION：每个分区的存储空间量，例如 30GiB

  • MESSAGE_RETENTION_PERIOD：精简版主题存储消息的时间，例如 1d 或 2w

  • NUMBER_OF_PARTITIONS：要为精简版主题配置的分区数。

如果请求成功，则响应是 JSON 格式的精简版主题：

{
  "name": projects/PROJECT_NUMBER/locations/LOCATION/topics/TOPIC_ID,
  "partitionConfig": {
      "count": NUMBER_OF_PARTITIONS,
      "capacity": {
         "publishMibPerSec": publish_CAPACITY,
         "subscribeMibPerSec": subscribe_CAPACITY,
      }
   },
   "retentionConfig": {
       "perPartitionBytes": STORAGE_PER_PARTITION,
       "period": MESSAGE_RETENTION_PERIOD,
   },
}

Go

运行此示例之前，请按照 Pub/Sub 精简版客户端库中的 Go 设置说明进行操作。

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

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

func updateTopic(w io.Writer, projectID, region, location, topicID, reservation string) error {
	// projectID := "my-project-id"
	// region := "us-central1"
	// NOTE: location can be either a region ("us-central1") or a zone ("us-central1-a")
	// For a list of valid locations, see https://cloud.google.com/pubsub/lite/docs/locations.
	// location := "us-central1"
	// topicID := "my-topic"
	// reservation := "projects/my-project-id/reservations/my-reservation"
	ctx := context.Background()
	client, err := pubsublite.NewAdminClient(ctx, region)
	if err != nil {
		return fmt.Errorf("pubsublite.NewAdminClient: %w", err)
	}
	defer client.Close()

	topicPath := fmt.Sprintf("projects/%s/locations/%s/topics/%s", projectID, location, topicID)
	// For ranges of fields in TopicConfigToUpdate, see https://pkg.go.dev/cloud.google.com/go/pubsublite/#TopicConfigToUpdate
	config := pubsublite.TopicConfigToUpdate{
		Name:                       topicPath,
		PartitionCount:             3, // Partition count cannot decrease.
		PublishCapacityMiBPerSec:   8,
		SubscribeCapacityMiBPerSec: 16,
		PerPartitionBytes:          60 * 1024 * 1024 * 1024,
		RetentionDuration:          24 * time.Hour,
		ThroughputReservation:      reservation,
	}
	updatedCfg, err := client.UpdateTopic(ctx, config)
	if err != nil {
		return fmt.Errorf("client.UpdateTopic got err: %w", err)
	}
	fmt.Fprintf(w, "Updated topic: %v\n", updatedCfg)
	return nil
}

Java

在运行此示例之前，请按照 Pub/Sub Lite 客户端库中的 Java 设置说明进行操作。

import com.google.api.gax.rpc.NotFoundException;
import com.google.cloud.pubsublite.AdminClient;
import com.google.cloud.pubsublite.AdminClientSettings;
import com.google.cloud.pubsublite.CloudRegion;
import com.google.cloud.pubsublite.CloudRegionOrZone;
import com.google.cloud.pubsublite.CloudZone;
import com.google.cloud.pubsublite.ProjectNumber;
import com.google.cloud.pubsublite.ReservationName;
import com.google.cloud.pubsublite.ReservationPath;
import com.google.cloud.pubsublite.TopicName;
import com.google.cloud.pubsublite.TopicPath;
import com.google.cloud.pubsublite.proto.Topic;
import com.google.cloud.pubsublite.proto.Topic.PartitionConfig;
import com.google.cloud.pubsublite.proto.Topic.PartitionConfig.Capacity;
import com.google.cloud.pubsublite.proto.Topic.ReservationConfig;
import com.google.cloud.pubsublite.proto.Topic.RetentionConfig;
import com.google.protobuf.FieldMask;
import com.google.protobuf.util.Durations;
import java.util.Arrays;
import java.util.concurrent.ExecutionException;

public class UpdateTopicExample {

  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';
    String topicId = "your-topic-id";
    String reservationId = "your-reservation-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 = true;

    updateTopicExample(cloudRegion, zoneId, projectNumber, topicId, reservationId, regional);
  }

  public static void updateTopicExample(
      String cloudRegion,
      char zoneId,
      long projectNumber,
      String topicId,
      String reservationId,
      boolean regional)
      throws Exception {

    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();

    ReservationPath reservationPath =
        ReservationPath.newBuilder()
            .setProject(ProjectNumber.of(projectNumber))
            .setLocation(CloudRegion.of(cloudRegion))
            .setName(ReservationName.of(reservationId))
            .build();

    Iterable<String> iterablePaths =
        Arrays.asList(
            "partition_config.scale",
            "retention_config.per_partition_bytes",
            "retention_config.period",
            "reservation_config.throughput_reservation");

    FieldMask fieldMask = FieldMask.newBuilder().addAllPaths(iterablePaths).build();

    Topic topic =
        Topic.newBuilder()
            .setPartitionConfig(
                PartitionConfig.newBuilder()
                    .setCapacity(
                        Capacity.newBuilder()
                            .setPublishMibPerSec(16)
                            .setSubscribeMibPerSec(32)
                            .build())
                    .build())
            .setRetentionConfig(
                RetentionConfig.newBuilder()
                    // Set storage per partition to 32 GiB. This must be 30 GiB-10 TiB.
                    // If the number of bytes stored in any of the topic's partitions grows
                    // beyond this value, older messages will be dropped to make room for
                    // newer ones, regardless of the value of `period`.
                    // Be careful when decreasing storage per partition as it may cause
                    // lost messages.
                    .setPerPartitionBytes(32 * 1024 * 1024 * 1024L)
                    .setPeriod(Durations.fromDays(7)))
            .setReservationConfig(
                ReservationConfig.newBuilder()
                    .setThroughputReservation(reservationPath.toString())
                    .build())
            .setName(topicPath.toString())
            .build();

    AdminClientSettings adminClientSettings =
        AdminClientSettings.newBuilder().setRegion(CloudRegion.of(cloudRegion)).build();

    try (AdminClient adminClient = AdminClient.create(adminClientSettings)) {
      Topic topicBeforeUpdate = adminClient.getTopic(topicPath).get();
      System.out.println("Before update: " + topicBeforeUpdate.getAllFields());

      Topic topicAfterUpdate = adminClient.updateTopic(topic, fieldMask).get();
      System.out.println("After update: " + topicAfterUpdate.getAllFields());
    } catch (ExecutionException e) {
      try {
        throw e.getCause();
      } catch (NotFoundException notFound) {
        System.out.println("This topic is not found.");
      } catch (Throwable throwable) {
        throwable.printStackTrace();
      }
    }
  }
}

Python

在运行此示例之前，请按照 Pub/Sub Lite 客户端库中的 Python 设置说明进行操作。

from google.api_core.exceptions import NotFound
from google.cloud.pubsublite import AdminClient, Topic
from google.cloud.pubsublite.types import (
    CloudRegion,
    CloudZone,
    ReservationPath,
    TopicPath,
)
from google.protobuf.duration_pb2 import Duration
from google.protobuf.field_mask_pb2 import FieldMask

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

location = None
if regional:
    #  A region.
    location = CloudRegion(cloud_region)
else:
    #  A zone.
    location = CloudZone(CloudRegion(cloud_region), zone_id)

topic_path = TopicPath(project_number, location, topic_id)
reservation_path = ReservationPath(project_number, cloud_region, reservation_id)

# Defines which topic fields to update.
field_mask = FieldMask(
    paths=[
        "partition_config.capacity",
        "retention_config.per_partition_bytes",
        "retention_config.period",
        "reservation_confing.throughput_reservation",
    ]
)

# Defines how to update the topic fields.
topic = Topic(
    name=str(topic_path),
    partition_config=Topic.PartitionConfig(
        capacity=Topic.PartitionConfig.Capacity(
            publish_mib_per_sec=16,
            subscribe_mib_per_sec=32,
        )
    ),
    retention_config=Topic.RetentionConfig(
        # Set storage per partition to 32 GiB. This must be in the range 30 GiB-10TiB.
        # If the number of byptes stored in any of the topic's partitions grows beyond
        # this value, older messages will be dropped to make room for newer ones,
        # regardless of the value of `period`.
        # Be careful when decreasing storage per partition as it may cuase lost messages.
        per_partition_bytes=32 * 1024 * 1024 * 1024,
        # Allow messages to be stored for 14 days.
        period=Duration(seconds=60 * 60 * 24 * 14),
    ),
    reservation_config=Topic.ReservationConfig(
        throughput_reservation=str(reservation_path),
    ),
)

client = AdminClient(cloud_region)
try:
    response = client.update_topic(topic, field_mask)
    print(f"{response.name} updated successfully.")
except NotFound:
    print(f"{topic_path} not found.")

查看精简版主题详情

您可以使用 Google Cloud 控制台、Google Cloud CLI 或 Pub/Sub Lite API 获取有关精简版主题的详细信息。

控制台

  1. 前往精简版主题页面。

    前往精简版主题

  2. 点击精简版主题 ID。

gcloud

如需获取有关精简版主题的详细信息，请使用 gcloud pubsub lite-topics describe 命令：

gcloud pubsub lite-topics describe TOPIC_ID \
--location=LOCATION

替换以下内容：

  • TOPIC_ID：精简版主题的 ID

  • LOCATION：精简版主题所在的位置名称

如果请求成功，命令行将显示精简版主题：

name: projects/PROJECT_NUMBER/locations/LOCATION/topics/TOPIC_ID
partitionConfig:
  count: NUMBER_OF_PARTITIONS
  capacity:
    publishMibPerSec: publish_CAPACITY
    subscribeMibPerSec: subscribe_CAPACITY
retentionConfig:
  perPartitionBytes: STORAGE_PER_PARTITION
  period: MESSAGE_RETENTION_PERIOD

协议

如需获取有关精简版主题的详细信息，请发送 GET 请求，如下所示：

GET https://REGION-pubsublite.googleapis.com/v1/admin/projects/PROJECT_NUMBER/locations/LOCATION/topics/TOPIC_ID
Authorization: Bearer $(gcloud auth print-access-token)

替换以下内容：

  • REGION：精简版主题所在的区域

  • PROJECT_NUMBER：具有精简版主题的项目的项目编号

  • LOCATION：精简版主题所在的位置名称

  • TOPIC_ID：精简版主题的 ID

如果请求成功，则响应是 JSON 格式的精简版主题：

{
  "name": projects/PROJECT_NUMBER/locations/LOCATION/topics/TOPIC_ID,
  "partitionConfig": {
      "count": NUMBER_OF_PARTITIONS,
      "capacity": {
         "publishMibPerSec": publish_CAPACITY,
         "subscribeMibPerSec": subscribe_CAPACITY,
      }
   },
   "retentionConfig": {
       "perPartitionBytes": STORAGE_PER_PARTITION,
       "period": MESSAGE_RETENTION_PERIOD,
   },
}

Go

运行此示例之前，请按照 Pub/Sub 精简版客户端库中的 Go 设置说明进行操作。

import (
	"context"
	"fmt"
	"io"

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

func getTopic(w io.Writer, projectID, region, location, topicID string) error {
	// projectID := "my-project-id"
	// region := "us-central1"
	// NOTE: location can be either a region ("us-central1") or a zone ("us-central1-a")
	// For a list of valid locations, see https://cloud.google.com/pubsub/lite/docs/locations.
	// location := "us-central1"
	// topicID := "my-topic"
	ctx := context.Background()
	client, err := pubsublite.NewAdminClient(ctx, region)
	if err != nil {
		return fmt.Errorf("pubsublite.NewAdminClient: %w", err)
	}
	defer client.Close()

	topicPath := fmt.Sprintf("projects/%s/locations/%s/topics/%s", projectID, location, topicID)
	topic, err := client.Topic(ctx, topicPath)
	if err != nil {
		return fmt.Errorf("client.Topic got err: %w", err)
	}
	fmt.Fprintf(w, "Got topic: %#v\n", *topic)
	return nil
}

Java

在运行此示例之前，请按照 Pub/Sub Lite 客户端库中的 Java 设置说明进行操作。

import com.google.api.gax.rpc.NotFoundException;
import com.google.cloud.pubsublite.AdminClient;
import com.google.cloud.pubsublite.AdminClientSettings;
import com.google.cloud.pubsublite.CloudRegion;
import com.google.cloud.pubsublite.CloudRegionOrZone;
import com.google.cloud.pubsublite.CloudZone;
import com.google.cloud.pubsublite.ProjectNumber;
import com.google.cloud.pubsublite.TopicName;
import com.google.cloud.pubsublite.TopicPath;
import com.google.cloud.pubsublite.proto.Topic;
import java.util.concurrent.ExecutionException;

public class GetTopicExample {

  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.
    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 = true;

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

  public static void getTopicExample(
      String cloudRegion, char zoneId, long projectNumber, String topicId, boolean regional)
      throws Exception {

    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();

    AdminClientSettings adminClientSettings =
        AdminClientSettings.newBuilder().setRegion(CloudRegion.of(cloudRegion)).build();

    try (AdminClient adminClient = AdminClient.create(adminClientSettings)) {
      Topic topic = adminClient.getTopic(topicPath).get();
      long numPartitions = adminClient.getTopicPartitionCount(topicPath).get();
      System.out.println(topic.getAllFields() + "\nhas " + numPartitions + " partition(s).");
    } catch (ExecutionException e) {
      try {
        throw e.getCause();
      } catch (NotFoundException notFound) {
        System.out.println("This topic is not found.");
      } catch (Throwable throwable) {
        throwable.printStackTrace();
      }
    }
  }
}

Python

在运行此示例之前，请按照 Pub/Sub Lite 客户端库中的 Python 设置说明进行操作。

from google.api_core.exceptions import NotFound
from google.cloud.pubsublite import AdminClient
from google.cloud.pubsublite.types import CloudRegion, CloudZone, TopicPath

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

location = None
if regional:
    #  A region.
    location = CloudRegion(cloud_region)
else:
    #  A zone.
    location = CloudZone(CloudRegion(cloud_region), zone_id)

topic_path = TopicPath(project_number, location, topic_id)

client = AdminClient(cloud_region)
try:
    response = client.get_topic(topic_path)
    num_partitions = client.get_topic_partition_count(topic_path)
    print(f"{response.name} has {num_partitions} partition(s).")
except NotFound:
    print(f"{topic_path} not found.")

列出精简版主题

您可以使用 Google Cloud 控制台、Google Cloud CLI 或 Pub/Sub Lite API 列出项目中的精简版主题。

控制台

  • 如需查看项目中的精简版主题列表，请前往精简版主题页面。

    前往精简版主题

gcloud

gcloud pubsub lite-topics list \
  --location=LOCATION

将 LOCATION 替换为精简版主题的位置名称。

如果请求成功，命令行将显示精简版主题：

---
name: projects/PROJECT_NUMBER/locations/LOCATION/topics/TOPIC_ID
partitionConfig:
  count: NUMBER_OF_PARTITIONS
  capacity:
    publishMibPerSec: publish_CAPACITY
    subscribeMibPerSec: subscribe_CAPACITY
retentionConfig:
  perPartitionBytes: STORAGE_PER_PARTITION
  period: MESSAGE_RETENTION_PERIOND
---
name: projects/PROJECT_NUMBER/locations/LOCATION/topics/TOPIC_ID
partitionConfig:
  count: NUMBER_OF_PARTITIONS
  capacity:
    publishMibPerSec: publish_CAPACITY
    subscribeMibPerSec: subscribe_CAPACITY
retentionConfig:
  perPartitionBytes: STORAGE_PER_PARTITION
  period: MESSAGE_RETENTION_PERIOND

协议

如需列出项目中的精简版主题，请发送 GET 请求，如下所示：

GET https://REGION-pubsublite.googleapis.com/v1/admin/projects/PROJECT_NUMBER/topics
Authorization: Bearer $(gcloud auth print-access-token)

替换以下内容：

  • REGION：精简版主题所在的区域

  • PROJECT_NUMBER：具有精简版主题的项目的项目编号

如果请求成功，则响应为 JSON 格式的精简版主题列表：

{
  "topics": [
      {
          "name": "projects/PROJECT_NUMBER/locations/LOCATION/topics/TOPIC_ID",
      },
      {
          "name": "projects/PROJECT_NUMBER/locations/LOCATION/topics/TOPIC_ID",
      }
  ]
}

Go

运行此示例之前，请按照 Pub/Sub 精简版客户端库中的 Go 设置说明进行操作。

import (
	"context"
	"fmt"
	"io"

	"cloud.google.com/go/pubsublite"
	"google.golang.org/api/iterator"
)

func listTopics(w io.Writer, projectID, region, zone string) error {
	// projectID := "my-project-id"
	// region := "us-central1"
	// zone := "us-central1-a"
	ctx := context.Background()
	client, err := pubsublite.NewAdminClient(ctx, region)
	if err != nil {
		return fmt.Errorf("pubsublite.NewAdminClient: %w", err)
	}
	defer client.Close()

	parent := fmt.Sprintf("projects/%s/locations/%s", projectID, zone)
	topicIter := client.Topics(ctx, parent)
	for {
		topic, err := topicIter.Next()
		if err == iterator.Done {
			break
		}
		if err != nil {
			return fmt.Errorf("topicIter.Next got err: %w", err)
		}
		fmt.Fprintf(w, "Got topic: %#v\n", topic)
	}
	return nil
}

Java

在运行此示例之前，请按照 Pub/Sub Lite 客户端库中的 Java 设置说明进行操作。

import com.google.cloud.pubsublite.AdminClient;
import com.google.cloud.pubsublite.AdminClientSettings;
import com.google.cloud.pubsublite.CloudRegion;
import com.google.cloud.pubsublite.CloudRegionOrZone;
import com.google.cloud.pubsublite.CloudZone;
import com.google.cloud.pubsublite.LocationPath;
import com.google.cloud.pubsublite.ProjectNumber;
import com.google.cloud.pubsublite.proto.Topic;
import java.util.List;

public class ListTopicsExample {

  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';
    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 = true;

    listTopicsExample(cloudRegion, zoneId, projectNumber, regional);
  }

  public static void listTopicsExample(
      String cloudRegion, char zoneId, long projectNumber, boolean regional) throws Exception {

    AdminClientSettings adminClientSettings =
        AdminClientSettings.newBuilder().setRegion(CloudRegion.of(cloudRegion)).build();

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

    LocationPath locationPath =
        LocationPath.newBuilder()
            .setProject(ProjectNumber.of(projectNumber))
            .setLocation(location)
            .build();

    try (AdminClient adminClient = AdminClient.create(adminClientSettings)) {
      List<Topic> topics = adminClient.listTopics(locationPath).get();
      for (Topic topic : topics) {
        System.out.println(topic.getAllFields());
      }
      System.out.println(topics.size() + " topic(s) listed.");
    }
  }
}

Python

在运行此示例之前，请按照 Pub/Sub Lite 客户端库中的 Python 设置说明进行操作。

from google.cloud.pubsublite import AdminClient
from google.cloud.pubsublite.types import CloudRegion, CloudZone, LocationPath

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

location = None
if regional:
    #  A region.
    location = CloudRegion(cloud_region)
else:
    #  A zone.
    location = CloudZone(CloudRegion(cloud_region), zone_id)

location_path = LocationPath(project_number, location)

client = AdminClient(cloud_region)
response = client.list_topics(location_path)

for topic in response:
    print(topic)

print(f"{len(response)} topic(s) listed in your project and location.")

删除精简版主题

您可以使用 Google Cloud 控制台、Google Cloud CLI 或 Pub/Sub Lite API 删除精简版主题。

控制台

  1. 前往精简版主题页面。

    前往精简版主题

  2. 点击精简版主题 ID。

  3. 在精简版主题详情页面中，点击删除。

  4. 在显示的字段中，输入 delete 以确认您想要删除精简版主题。

  5. 点击删除。

gcloud

要删除精简版主题，请使用 gcloud pubsub lite-topics delete 命令：

  1. 运行 delete 命令：

    gcloud pubsub lite-topics delete TOPIC_ID \
 --location=LOCATION

    替换以下内容：

    • TOPIC_ID：精简版主题的 ID

    • LOCATION：精简版主题所在的位置名称

  2. 如需确认，请输入 Y。

如果请求成功，则响应为以下内容：

Deleted topic [TOPIC_ID].

协议

要删除精简版主题，请发送 DELETE 请求，如下所示：

DELETE https://REGION-pubsublite.googleapis.com/v1/admin/projects/PROJECT_NUMBER/locations/LOCATION/topics/TOPIC_ID
Authorization: Bearer $(gcloud auth print-access-token)

替换以下内容：

  • REGION：精简版主题所在的区域

  • PROJECT_NUMBER：具有精简版主题的项目的项目编号

  • LOCATION：精简版主题所在的位置名称

  • TOPIC_ID：精简版主题的 ID

如果请求成功，响应将为空的 JSON 对象。

Go

运行此示例之前，请按照 Pub/Sub 精简版客户端库中的 Go 设置说明进行操作。

import (
	"context"
	"fmt"
	"io"

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

func deleteTopic(w io.Writer, projectID, region, location, topicID string) error {
	// projectID := "my-project-id"
	// region := "us-central1"
	// NOTE: location can be either a region ("us-central1") or a zone ("us-central1-a")
	// For a list of valid locations, see https://cloud.google.com/pubsub/lite/docs/locations.
	// location := "us-central1"
	// topicID := "my-topic"
	ctx := context.Background()
	client, err := pubsublite.NewAdminClient(ctx, region)
	if err != nil {
		return fmt.Errorf("pubsublite.NewAdminClient: %w", err)
	}
	defer client.Close()

	topicPath := fmt.Sprintf("projects/%s/locations/%s/topics/%s", projectID, location, topicID)
	err = client.DeleteTopic(ctx, topicPath)
	if err != nil {
		return fmt.Errorf("client.DeleteTopic got err: %w", err)
	}
	fmt.Fprint(w, "Deleted topic\n")
	return nil
}

Java

在运行此示例之前，请按照 Pub/Sub Lite 客户端库中的 Java 设置说明进行操作。

import com.google.api.gax.rpc.NotFoundException;
import com.google.cloud.pubsublite.AdminClient;
import com.google.cloud.pubsublite.AdminClientSettings;
import com.google.cloud.pubsublite.CloudRegion;
import com.google.cloud.pubsublite.CloudRegionOrZone;
import com.google.cloud.pubsublite.CloudZone;
import com.google.cloud.pubsublite.ProjectNumber;
import com.google.cloud.pubsublite.TopicName;
import com.google.cloud.pubsublite.TopicPath;
import java.util.concurrent.ExecutionException;

public class DeleteTopicExample {

  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.
    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;

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

  public static void deleteTopicExample(
      String cloudRegion, char zoneId, long projectNumber, String topicId, boolean regional)
      throws Exception {

    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();

    AdminClientSettings adminClientSettings =
        AdminClientSettings.newBuilder().setRegion(CloudRegion.of(cloudRegion)).build();

    try (AdminClient adminClient = AdminClient.create(adminClientSettings)) {
      adminClient.deleteTopic(topicPath).get();
      if (regional) {
        System.out.println(topicPath.toString() + " (regional topic) deleted successfully.");
      } else {
        System.out.println(topicPath.toString() + " (zonal topic) deleted successfully.");
      }
    } catch (ExecutionException e) {
      try {
        throw e.getCause();
      } catch (NotFoundException notFound) {
        System.out.println("This topic is not found.");
      } catch (Throwable throwable) {
        throwable.printStackTrace();
      }
    }
  }

Python

在运行此示例之前，请按照 Pub/Sub Lite 客户端库中的 Python 设置说明进行操作。

from google.api_core.exceptions import NotFound
from google.cloud.pubsublite import AdminClient
from google.cloud.pubsublite.types import CloudRegion, CloudZone, TopicPath

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

cloud_region = CloudRegion(cloud_region)
topic_path = None
if regional:
    #  A regional topic.
    topic_path = TopicPath(project_number, cloud_region, topic_id)
else:
    #  A zonal topic
    topic_path = TopicPath(
        project_number, CloudZone(cloud_region, zone_id), topic_id
    )

client = AdminClient(cloud_region)
try:
    client.delete_topic(topic_path)
    if regional:
        print(f"{topic_path} (regional topic) deleted successfully.")
    else:
        print(f"{topic_path} (zonal topic) deleted successfully.")
except NotFound:
    print(f"{topic_path} not found.")

如果删除精简版主题，则无法向其发布消息。精简版主题仍然存在，但您无法从精简版订阅接收消息。

后续步骤