创建和管理精简版主题

本页面介绍如何创建、查看和删除精简版主题。

精简版主题是您可以向其发布消息的区域资源。创建精简版主题时,您必须指定用于存储消息的位置。

创建精简版主题后,您可以向精简版主题发布消息创建精简版订阅以及从精简版订阅接收消息

创建精简版主题

创建精简版主题时,必须设置其吞吐量和存储容量。分区的数量和容量决定了精简版主题的容量。

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

  • 分区数:精简版主题中的分区数。

  • 吞吐量预留预留是预配和管理吞吐量容量的最简单方法。

  • 发布吞吐量容量:如果您不指定预留,则必须在主题中明确预配每个分区的发布和订阅容量。您可以指定每秒 4 到 16 MiB 的吞吐量容量,且值必须是整数。项目会为此吞吐量容量付费。如果您指定了预留,则可以使用此属性来限制主题的吞吐量。在这种情况下,容量下限不适用,您不需要为该主题单独付费。

  • 订阅吞吐量容量:每个分区的最大订阅吞吐量。您可以指定每秒 4 到 32 MiB 的吞吐量容量,并且值必须是整数。

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

  • 存储空间配置

    • 每个分区的存储空间:每个分区的存储空间量(以字节为单位)。您可以指定 30 GiB - 10 TiB 的存储空间。
    • 消息保留期限:精简版主题存储消息的最长时间。如果您未指定消息保留期限,Lite 主题会存储消息,直至超出存储容量。

您可以使用 Cloud Console、gcloud 命令行工具或 Pub/Sub Lite API 创建精简版主题。

控制台

  1. 在 Cloud Console 中,转到精简版主题页面。

    转到精简版主题页面

  2. 点击创建精简版主题

  3. 选择一个地区和该地区中的区域。

  4. 名称部分中,输入精简版主题 ID。精简版主题名称包括精简版主题 ID、位置和项目编号。

  5. 吞吐量部分中,输入分区数以预配分区。

  6. 消息存储部分中,输入每个分区的存储空间

  7. 可选:在消息保留部分,选择即使存储空间可用,也要在指定的期限后删除消息,然后输入保留期限。

  8. 点击创建

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 \
  [--message-retention-period=MESSAGE_RETENTION_PERIOD]

替换以下内容:

  • TOPIC_ID:精简版主题的 ID

  • LOCATIONPub/Sub 精简版支持的位置的名称

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

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

  • MESSAGE_RETENTION_PERIOD:精简版主题存储消息的时间,例如 1d2w

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

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": PUBLISHING_CAPACITY,
         "subscribeMibPerSec": SUBSCRIBING_CAPACITY,
       }
  },
  "retentionConfig": {
       "perPartitionBytes": STORAGE_PER_PARTITION,
       "period": MESSAGE_RETENTION_PERIOD,
  },
}

替换以下内容:

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

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

  • PUBLISHING_CAPACITY:一个整数,表示每个分区的发布吞吐量容量

  • SUBSCRIBING_CAPACITY:一个整数,表示每个分区的订阅吞吐量容量

  • MESSAGE_RETENTION_PERIOD:精简版主题存储消息的时间,例如 1d2w

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

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

Go

在尝试此示例之前,请按照《快速入门:使用客户端库》中的 Go 设置说明进行操作。 如需了解详情,请参阅 Pub/Sub Go API 参考文档

import (
	"context"
	"fmt"
	"io"

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

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

	const gib = 1 << 30
	// 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:                       fmt.Sprintf("projects/%s/locations/%s/topics/%s", projectID, zone, topicID),
		PartitionCount:             2, // Must be >= 1 and cannot decrease after creation.
		PublishCapacityMiBPerSec:   4,
		SubscribeCapacityMiBPerSec: 8,
		PerPartitionBytes:          30 * gib,
		RetentionDuration:          pubsublite.InfiniteRetention,
	})
	if err != nil {
		return fmt.Errorf("client.CreateTopic got err: %v", err)
	}
	fmt.Fprintf(w, "Created topic: %s\n", topic.Name)
	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.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 com.google.cloud.pubsublite.proto.Topic.PartitionConfig;
import com.google.cloud.pubsublite.proto.Topic.RetentionConfig;
import com.google.protobuf.util.Durations;

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 = 'b';
    String topicId = "your-topic-id";
    long projectNumber = Long.parseLong("123456789");
    int partitions = 1;

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

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

    TopicPath topicPath =
        TopicPath.newBuilder()
            .setProject(ProjectNumber.of(projectNumber))
            .setLocation(CloudZone.of(CloudRegion.of(cloudRegion), zoneId))
            .setName(TopicName.of(topicId))
            .build();

    Topic topic =
        Topic.newBuilder()
            .setPartitionConfig(
                PartitionConfig.newBuilder()
                    // Set throughput capacity per partition in MiB/s.
                    .setCapacity(
                        PartitionConfig.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))
            .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();
      System.out.println(response.getAllFields() + "created successfully.");
    }
  }
}

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, 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"
# num_partitions = 1

cloud_region = CloudRegion(cloud_region)
location = CloudZone(cloud_region, zone_id)
topic_path = TopicPath(project_number, location, 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),
    ),
)

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

如需查看可用地区的列表,请参阅 Pub/Sub 精简版位置

创建精简版主题后,您可以扩缩每个分区的吞吐量和存储空间容量。您也可以增加分区数,但不能减少。如需了解详情,请参阅扩缩容量

预配容量

精简版主题由分区组成。分区支持以下容量:

  • 发布吞吐量:您可以发布消息的最大总速率。
  • 订阅吞吐量:将消息转发到精简版订阅的最大总速率。
  • 存储:分区中消息的最大总大小。

如果超出存储空间容量,Pub/Sub Lite 服务不会发布失败。而是会从分区中移除最早的消息,无论最早的消息的消息保留期限如何。Pub/Sub 精简版服务会尽最大努力移除消息。

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

更新精简版主题

您可以更新精简版主题的消息保留期限和容量。您无法更新主题的可用区。要扩缩精简版主题的容量,请参阅扩缩容量

您可以使用 Cloud Console、gcloud 命令行工具或 Pub/Sub Lite API 更新精简版主题。

控制台

  1. 在 Cloud Console 中,转到精简版主题页面。

    转到精简版主题页面

  2. 点击精简版主题 ID。

  3. 点击修改

  4. 输入分区数。这可能会增加分区数,但不能降低。

  5. 输入每个分区的存储空间保留期限

  6. 选择发布吞吐量

  7. 选择订阅吞吐量

  8. 输入每个分区的存储

gcloud

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

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

替换以下内容:

  • TOPIC_ID:精简版主题的 ID

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

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

  • PUBLISHING_CAPACITY:一个整数,表示每个分区的发布吞吐量容量

  • SUBSCRIBING_CAPACITY:一个整数,表示每个分区的订阅吞吐量容量

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

  • MESSAGE_RETENTION_PERIOD:精简版主题存储消息的时间,例如 1d2w

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

name: projects/PROJECT_NUMBER/locations/LOCATION/topics/TOPIC_ID
partitionConfig:
  count: NUMBER_OF_PARTITIONS
  capacity:
    publishMibPerSec: PUBLISHING_CAPACITY
    subscribeMibPerSec: SUBSCRIBING_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": PUBLISHING_CAPACITY,
         "subscribeMibPerSec": SUBSCRIBING_CAPACITY,
      }
   },
   "retentionConfig": {
       "perPartitionBytes": STORAGE_PER_PARTITION,
       "period": MESSAGE_RETENTION_PERIOD,
   },
}

替换以下内容:

  • PUBLISHING_CAPACITY:一个整数,表示每个分区的发布吞吐量容量

  • SUBSCRIBING_CAPACITY:一个整数,表示每个分区的订阅吞吐量容量

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

  • MESSAGE_RETENTION_PERIOD:精简版主题存储消息的时间,例如 1d2w

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

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

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

Go

在尝试此示例之前,请按照《快速入门:使用客户端库》中的 Go 设置说明进行操作。 如需了解详情,请参阅 Pub/Sub Go API 参考文档

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

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

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

	topicPath := fmt.Sprintf("projects/%s/locations/%s/topics/%s", projectID, zone, 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,
	}
	updatedCfg, err := client.UpdateTopic(ctx, config)
	if err != nil {
		return fmt.Errorf("client.UpdateTopic got err: %v", err)
	}
	fmt.Fprintf(w, "Updated topic: %#v\n", *updatedCfg)
	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.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 com.google.cloud.pubsublite.proto.Topic.PartitionConfig;
import com.google.cloud.pubsublite.proto.Topic.RetentionConfig;
import com.google.protobuf.FieldMask;
import com.google.protobuf.util.Durations;
import java.util.Arrays;

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";
    long projectNumber = Long.parseLong("123456789");

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

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

    TopicPath topicPath =
        TopicPath.newBuilder()
            .setProject(ProjectNumber.of(projectNumber))
            .setLocation(CloudZone.of(CloudRegion.of(cloudRegion), zoneId))
            .setName(TopicName.of(topicId))
            .build();

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

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

    Topic topic =
        Topic.newBuilder()
            .setPartitionConfig(
                PartitionConfig.newBuilder()
                    // Set publishing throughput to 4 times the standard partition
                    // throughput of 4 MiB per sec. This must be in the range [1,4]. A
                    // topic with `scale` of 2 and count of 10 is charged for 20 partitions.
                    .setScale(4)
                    .build())
            .setRetentionConfig(
                RetentionConfig.newBuilder()
                    // Set storage per partition to 200 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(200 * 1024 * 1024 * 1024L)
                    .setPeriod(Durations.fromDays(7)))
            .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());
    }
  }
}

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, 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"

cloud_region = CloudRegion(cloud_region)
location = CloudZone(cloud_region, zone_id)
topic_path = TopicPath(project_number, location, topic_id)

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

# Defines how to update the topic fields.
topic = Topic(
    name=str(topic_path),
    partition_config=Topic.PartitionConfig(
        # Set publishing throughput to 2x standard partition throughput of 4 MiB
        # per second. This must in the range [1,4]. A topic with `scale` of 2 and
        # `count` of 10 is charged for 20 partitions.
        scale=2,
    ),
    retention_config=Topic.RetentionConfig(
        # Set storage per partition to 100 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=100 * 1024 * 1024 * 1024,
        # Allow messages to be stored for 14 days.
        period=Duration(seconds=60 * 60 * 24 * 14),
    ),
)

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.")

扩缩吞吐量容量

您可以纵向和横向扩缩主题的吞吐量容量。纵向扩缩包括增加或减少主题分区的容量。Pub/Sub 精简版服务为精简版主题中的每个分区预配了相同数量的吞吐量容量。例如,如果您将发布吞吐量容量增加到每秒 10 MiB,则每个分区以每秒 10 MiB 发布吞吐量。

您还可以通过增加分区数量在水平上增加精简版主题的容量。主题中的分区数不能减少。

当您更改分区数时,系统不会保留消息的相对顺序。更新主题中的分区数会更改用于从排序键映射到分区的哈希函数。分区数更新后发布的具有相同键的消息可能会映射到与更新之前发布的消息所在分区不同的分区。此外,主题调整大小功能需要花费一段时间才能传播到所有发布商,因此在短时间内有些发布商使用新哈希函数,而有些发布商仍使用旧哈希函数。

扩缩存储空间容量

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

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

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

获取精简版主题详情

您可以使用 Cloud Console、gcloud 命令行工具或 Pub/Sub Lite API 获取有关精简版主题的详细信息。

控制台

  1. 在 Cloud Console 中,转到精简版主题页面。

    转到精简版主题页面

  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: PUBLISHING_CAPACITY
    subscribeMibPerSec: SUBSCRIBING_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": PUBLISHING_CAPACITY,
         "subscribeMibPerSec": SUBSCRIBING_CAPACITY,
      }
   },
   "retentionConfig": {
       "perPartitionBytes": STORAGE_PER_PARTITION,
       "period": MESSAGE_RETENTION_PERIOD,
   },
}

Go

在尝试此示例之前,请按照《快速入门:使用客户端库》中的 Go 设置说明进行操作。 如需了解详情,请参阅 Pub/Sub Go API 参考文档

import (
	"context"
	"fmt"
	"io"

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

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

	topicName := fmt.Sprintf("projects/%s/locations/%s/topics/%s", projectID, zone, topicID)
	topic, err := client.Topic(ctx, topicName)
	if err != nil {
		return fmt.Errorf("client.Topic got err: %v", 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.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;

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

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

  public static void getTopicExample(
      String cloudRegion, char zoneId, long projectNumber, String topicId) throws Exception {
    TopicPath topicPath =
        TopicPath.newBuilder()
            .setProject(ProjectNumber.of(projectNumber))
            .setLocation(CloudZone.of(CloudRegion.of(cloudRegion), zoneId))
            .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).");
    }
  }
}

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"

cloud_region = CloudRegion(cloud_region)
location = CloudZone(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.")

列出精简版主题

您可以使用 Cloud Console、gcloud 命令行工具或 Pub/Sub Lite API 列出项目中的精简版主题。

控制台

如需查看项目中的精简版主题列表,请转到精简版主题页面

gcloud

如需列出项目中的精简版主题,请使用 gcloud pubsub lite-topics list 命令:

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

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

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

---
name: projects/PROJECT_NUMBER/locations/LOCATION/topics/TOPIC_ID
partitionConfig:
  count: NUMBER_OF_PARTITIONS
  capacity:
    publishMibPerSec: PUBLISHING_CAPACITY
    subscribeMibPerSec: SUBSCRIBING_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: PUBLISHING_CAPACITY
    subscribeMibPerSec: SUBSCRIBING_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

在尝试此示例之前,请按照《快速入门:使用客户端库》中的 Go 设置说明进行操作。 如需了解详情,请参阅 Pub/Sub Go API 参考文档

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: %v", 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: %v", 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.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");

    listTopicsExample(cloudRegion, zoneId, projectNumber);
  }

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

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

    LocationPath locationPath =
        LocationPath.newBuilder()
            .setProject(ProjectNumber.of(projectNumber))
            .setLocation(CloudZone.of(CloudRegion.of(cloudRegion), zoneId))
            .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"

cloud_region = CloudRegion(cloud_region)
location = CloudZone(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.")

删除精简版主题

您可以使用 Cloud Console、gcloud命令行工具或 Pub/Sub Lite API 删除精简版主题。

控制台

  1. 在 Cloud Console 中,转到精简版主题页面。

    转到精简版主题页面

  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

在尝试此示例之前,请按照《快速入门:使用客户端库》中的 Go 设置说明进行操作。 如需了解详情,请参阅 Pub/Sub Go API 参考文档

import (
	"context"
	"fmt"
	"io"

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

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

	err = client.DeleteTopic(ctx, fmt.Sprintf("projects/%s/locations/%s/topics/%s", projectID, zone, topicID))
	if err != nil {
		return fmt.Errorf("client.DeleteTopic got err: %v", err)
	}
	fmt.Fprint(w, "Deleted topic\n")
	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.CloudZone;
import com.google.cloud.pubsublite.ProjectNumber;
import com.google.cloud.pubsublite.TopicName;
import com.google.cloud.pubsublite.TopicPath;

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

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

  public static void deleteTopicExample(
      String cloudRegion, char zoneId, long projectNumber, String topicId) throws Exception {
    TopicPath topicPath =
        TopicPath.newBuilder()
            .setProject(ProjectNumber.of(projectNumber))
            .setLocation(CloudZone.of(CloudRegion.of(cloudRegion), zoneId))
            .setName(TopicName.of(topicId))
            .build();

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

    try (AdminClient adminClient = AdminClient.create(adminClientSettings)) {
      adminClient.deleteTopic(topicPath).get();
      System.out.println(topicPath.toString() + " deleted successfully.");
    }
  }

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"

cloud_region = CloudRegion(cloud_region)
location = CloudZone(cloud_region, zone_id)
topic_path = TopicPath(project_number, location, topic_id)

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

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