创建 Cloud Storage 订阅

本文档介绍了如何创建 Cloud Storage 订阅。 您可以使用 Google Cloud 控制台、Google Cloud CLI、客户端库或 Pub/Sub API 创建 Cloud Storage 订阅。

准备工作

在阅读本文档之前，请确保您熟悉以下内容：

• Cloud Storage 订阅的运作方式。
• Cloud Storage 的工作原理，以及如何创建和管理 Cloud Storage 存储桶。
• 如何配置死信主题来处理消息失败。

所需的角色和权限

如需获得创建 Cloud Storage 订阅所需的权限，请让您的管理员为您授予项目的 Pub/Sub Editor (roles/pubsub.editor) IAM 角色。 如需详细了解如何授予角色，请参阅管理对项目、文件夹和组织的访问权限。

此预定义角色包含创建 Cloud Storage 订阅所需的权限。如需查看所需的确切权限，请展开所需权限部分：

所需权限

如需创建 Cloud Storage 订阅，您需要具备以下权限：

• 针对项目的 pubsub.subscriptions.create 权限
• pubsub.topics.attachSubscription 主题

您也可以使用自定义角色或其他预定义角色来获取这些权限。

跨项目订阅

如果您在一个项目中为另一个项目中的主题创建订阅，则必须对创建订阅的项目拥有 pubsub.subscriptions.create 权限，并且对相应主题拥有 pubsub.topics.attachSubscription 权限。

向服务账号授予 IAM 角色

Pub/Sub 使用 Identity and Access Management (IAM) 服务账号访问 Google Cloud 资源。默认情况下，它使用 Pub/Sub 服务代理 (service-PROJECT_NUMBER@gcp-sa-pubsub.)。

如需让 Pub/Sub 能够写入 Cloud Storage，服务账号需要具备以下角色：

• Storage Object Creator (roles/storage.objectCreator)
• Storage Legacy Bucket Reader (roles/storage.legacyBucketReader)

您可以为项目或 Cloud Storage 存储桶授予服务账号权限，如下所示：

项目

1. 在 Google Cloud 控制台中，前往存储桶页面。

   进入“存储桶”

2. 选择包括 Google 提供的角色授权。

3. 找到 Cloud Pub/Sub 服务账号对应的行，然后点击 修改主账号。

4. 点击添加其他角色，然后选择 Storage Object Creator 角色。 对 Storage Legacy Bucket Reader 角色重复此步骤。

如需了解详情，请参阅使用控制台授予 IAM 角色。

Cloud Storage 存储桶

1. 在 Google Cloud 控制台中，前往存储桶。

   进入“存储桶”

2. 点击要向其授予权限的 Cloud Storage 存储桶的名称。

3. 在存储桶详情页面中，点击配置标签页。

4. 在权限窗格中，点击按主账号查看标签页。

5. 点击授予访问权限。

6. 在新的主账号字段中，输入服务账号标识符，格式如下：

   service-PROJECT_NUMBER@gcp-sa-pubsub.。

7. 在分配角色列表中，选择 Storage Object Creator。

8. 点击 添加其他角色，然后选择 Storage Legacy Bucket Reader。

9. 点击保存。系统会向主账号授予资源上的角色。

使用自定义服务账号

通过向 Cloud Pub/Sub 服务账号授予 Storage Object Creator 和 Storage Legacy Bucket Reader 角色，任何有权在项目中创建订阅的用户都可以写入 Cloud Storage 存储桶。如果您想提供更精细的权限，请改为配置用户代管式服务账号。

如需配置用户代管式服务账号以写入 Cloud Storage，您需要具有以下权限：

• 用户代管式服务账号必须具有 Storage Object Creator 和 Storage Legacy Bucket Reader 角色。

• Cloud Pub/Sub 服务账号必须对用户代管式服务账号具有 iam.serviceAccounts.getAccessToken 权限。

• 创建订阅的用户必须对用户代管式服务账号拥有 iam.serviceAccounts.actAs 权限。

创建订阅时，请将用户代管式服务账号指定为订阅服务账号。

Cloud Storage 订阅属性

Cloud Storage 订阅支持所有常见订阅属性。以下部分介绍了 Cloud Storage 订阅特有的属性。

存储桶名称

在创建 Cloud Storage 订阅之前，Cloud Storage 存储桶必须已存在。

这些消息会分批发送并存储在 Cloud Storage 存储桶中。单个批次或文件以对象的形式存储在存储桶中。

Cloud Storage 存储桶必须停用请求者付款。

如需创建 Cloud Storage 存储桶，请参阅创建存储桶。

文件名前缀、后缀和日期时间

由 Cloud Storage 订阅生成的输出 Cloud Storage 文件以对象的形式存储在 Cloud Storage 存储桶中。存储在 Cloud Storage 存储桶中的对象的名称采用以下格式：<file-prefix><UTC-date-time>_<uuid><file-suffix>。

以下列表包含文件格式和可自定义字段的详细信息：

• <file-prefix> 是自定义文件名前缀。此字段为选填字段。

• <UTC-date-time> 是根据对象创建时间自动生成的自定义字符串。

• <uuid> 是为对象自动生成的随机字符串。

• <file-suffix> 是自定义文件名后缀。此为可选字段。文件名后缀不能以“/”结尾。

• 您可以更改文件名前缀和后缀：

  • 例如，如果文件名前缀的值为 prod_，文件后缀的值为 _archive，则示例对象名称为 prod_2023-09-25T04:10:00+00:00_uN1QuE_archive。

  • 如果您未指定文件名前缀和后缀，则存储在 Cloud Storage 存储桶中的对象名称的格式为：<UTC-date-time>_<uuid>。

  • Cloud Storage 对象命名要求也适用于文件名前缀和后缀。如需了解详情，请参阅 Cloud Storage 对象简介。

• 您可以更改文件名中日期和时间的显示方式：

  • 必需的日期时间匹配器，只能使用一次：年（YYYY 或 YY）、月（MM）、日（DD）、时（hh）、分（mm）、秒（ss）。例如，YY-YYYY 或 MMM 无效。

  • 只能使用一次的可选匹配器：日期时间分隔符 (T) 和时区偏移量（Z 或 +00:00）。

  • 可多次使用的可选元素：连字符 (-)、下划线 (_)、英文冒号 (:) 和正斜杠 (/)。

  • 例如，如果文件名日期时间格式的值为 YYYY-MM-DD/hh_mm_ssZ，则示例对象名称为 prod_2023-09-25/04_10_00Z_uNiQuE_archive。

  • 如果文件名日期时间格式以非匹配器字符结尾，该字符将替换 <UTC-date-time> 和 <uuid> 之间的分隔符。例如，如果文件名日期时间格式的值为 YYYY-MM-DDThh_mm_ss-，则示例对象名称为 prod_2023-09-25T04_10_00-uNiQuE_archive。

文件批量处理

通过 Cloud Storage 订阅，您可以决定何时创建新的输出文件，该文件会作为对象存储在 Cloud Storage 存储桶中。当满足指定的分批条件之一时，Pub/Sub 会写入输出文件。以下是 Cloud Storage 批处理条件：

• 批量存储时长上限。这是一项必需设置。 如果超过指定的最大时长值，Pub/Sub 会写入新的输出文件。此时长是指从 Pub/Sub 开始写入新文件到文件最终确定之间的时间。例如，如果您将最长时长设置为 5 分钟，Pub/Sub 会在开始向文件写入数据后最多 5 分钟内完成该文件的写入。系统可能会在时长达到上限前创建新文件。如果您未指定值，系统会应用默认值 5 分钟。以下是最大时长的适用值：

  • 最小值 = 1 分钟
  • 默认值 = 5 分钟
  • 最大值 = 10 分钟

• 存储批量操作的字节数上限。这是一项可选设置。如果超出指定的最大字节数值，Cloud Storage 订阅会写入新的输出文件。以下是适用于最大字节数的值：

  • 最小值 = 1 KB
  • 最大值 = 10 GiB

• 批量存储消息数上限。这是一项可选设置。如果超过指定的消息数量上限，Cloud Storage 订阅会写入新的输出文件。以下是“最多消息数”的适用值：

  • 最小值 = 1000

例如，您可以将最长时长配置为 6 分钟，并将最大字节数配置为 2 GB。如果在第 4 分钟时，输出文件的大小达到 2 GB，Pub/Sub 会完成上一个文件的写入，并开始写入新文件。

Cloud Storage 订阅可能会同时写入 Cloud Storage 存储桶中的多个文件。如果您已将订阅配置为每 6 分钟创建一个新文件，则可能会发现每 6 分钟会创建多个 Cloud Storage 文件。

在某些情况下，Pub/Sub 可能会比文件批处理条件配置的时间更早开始写入新文件。如果订阅接收的消息大于“最大字节数”值，文件也可能会超出“最大字节数”值。

文件格式

创建 Cloud Storage 订阅时，您可以将要存储在 Cloud Storage 存储桶中的输出文件格式指定为 Text 或 Avro。

• 文本：消息以纯文本形式存储。换行符用于分隔消息与文件中的上一条消息。系统只会存储消息载荷，而不会存储属性或其他元数据。

• Avro：消息以 Apache Avro 二进制格式存储。选择 Avro 时，您可以启用以下其他属性：

  • 写入元数据：此选项可让您将消息元数据与消息一起存储。subscription_name、message_id、publish_time 和 attributes 字段等元数据会写入输出 Avro 对象中的顶级字段，而除数据之外的所有其他消息属性（例如 ordering_key，如果存在）都会作为条目添加到 attributes 映射中。

    如果停用了写入元数据，则只会将消息载荷写入输出 Avro 对象。以下是停用写入元数据的输出消息的 Avro 架构：

    {
      "type": "record",
      "namespace": "com.google.pubsub",
      "name": "PubsubMessage",
      "fields": [
        { "name": "data", "type": "bytes" }
      ]
    }
    

    以下是启用了写入元数据的输出消息的 Avro 架构：

    {
      "type": "record",
      "namespace": "com.google.pubsub",
      "name": "PubsubMessageWithMetadata",
      "fields": [
        { "name": "subscription_name", "type": "string" },
        { "name": "message_id", "type": "string"  },
        { "name": "publish_time", "type": {
            "type": "long",
            "logicalType": "timestamp-micros"
          }
        },
        { "name": "attributes", "type": { "type": "map", "values": "string" } },
        { "name": "data", "type": "bytes" }
      ]
    }
    

  • 使用主题架构：此选项可让 Pub/Sub 在写入 Avro 文件时使用订阅所附加的 Pub/Sub 主题的架构。

    使用此选项时，请务必检查是否满足以下附加要求：

    • 主题架构必须采用 Apache Avro 格式。

    • 如果使用主题架构和写入元数据均已启用，则主题架构的根目录中必须有一个 Record 对象。Pub/Sub 将展开记录的字段列表，以包含元数据字段。因此，Record 不能包含与元数据字段（subscription_name、message_id、publish_time 或 attributes）同名的任何字段。

注意： 对于有兴趣以 Parquet 格式写入文件的用户，将 Pub/Sub 数据流式传输到 Parquet 文件的最佳方式是使用 Apache Iceberg 的 BigQuery 表的 BigQuery 订阅。如需了解详情，请参阅使用 Iceberg 表的 BigQuery 订阅。

服务账号

您可以通过以下方式将消息写入 Cloud Storage 存储桶：

• 配置自定义服务账号，以便只有拥有该服务账号的 iam.serviceAccounts.actAs 权限的用户才能创建写入相应存储桶的订阅。Service Account User (roles/iam.serviceAccountUser) 角色是包含 iam.serviceAccounts.actAs 权限的一个示例角色。

• 使用默认 Pub/Sub 服务代理，该代理允许任何能够在项目中创建订阅的用户创建向存储桶写入数据的订阅。如果您未指定自定义服务账号，则 Pub/Sub 服务代理是默认设置。

创建 Cloud Storage 订阅

控制台

1. 在 Google Cloud 控制台中，前往订阅页面。

   前往“订阅”

2. 点击创建订阅。

3. 在订阅 ID 字段中，输入一个名称。

   如需了解如何命名订阅，请参阅主题或订阅命名指南。

4. 从下拉菜单中选择或创建一个主题。

   订阅将接收来自该主题的消息。

   如需了解如何创建主题，请参阅创建和管理主题。

5. 选择写入 Cloud Storage 作为传送类型。

6. 对于 Cloud Storage 存储桶，请点击浏览。

   • 您可以从任何合适的项目中选择现有存储桶。

   • 您也可以点击“创建”图标，然后按照屏幕上的说明创建新存储桶。

     创建存储桶后，为 Cloud Storage 订阅选择该存储桶。

     如需详细了解如何创建存储桶，请参阅创建存储桶。

   指定存储桶后，Pub/Sub 会检查 Pub/Sub 服务代理是否对该存储桶拥有适当的权限。如果存在权限问题，您会看到类似如下的消息：Unable to verify if the Pub/Sub service agent has write permissions on this bucket. You may be lacking permissions to view or set permissions。

7. 如果您遇到权限问题，请点击设置权限，然后按照屏幕上的说明操作。

   或者，按照向 Pub/Sub 服务代理分配 Cloud Storage 角色中的说明操作。

8. 对于文件格式，请选择文本或 Avro。

   如果您选择 Avro，还可以选择指定是否要在输出中存储消息元数据。

   如需详细了解这两个选项（包括 Avro 格式的消息元数据选项），请参阅文件格式。

9. 可选：您可以为要写入 Cloud Storage 存储桶的所有文件指定文件名前缀、后缀和日期时间。文件以对象的形式存储在存储桶中。

   如需详细了解如何设置文件前缀、后缀和日期时间，请参阅文件名、前缀、后缀和日期时间。

10. 对于文件批处理，请指定创建新文件之前经过的最长时间。

    您还可以选择性地设置文件的最大文件大小或最大消息数量。

    如需详细了解这两种文件批处理选项，请参阅文件批处理。

11. 我们强烈建议您启用 Dead lettering 来处理消息失败。

    如需了解详情，请参阅死信主题。

12. 您可以将其他设置保留为默认值，然后点击创建。

gcloud

1. 在 Google Cloud 控制台中，激活 Cloud Shell。

   激活 Cloud Shell

   Cloud Shell 会话随即会在 Google Cloud 控制台的底部启动，并显示命令行提示符。Cloud Shell 是一个已安装 Google Cloud CLI 且已为当前项目设置值的 Shell 环境。该会话可能需要几秒钟时间来完成初始化。

2. 如需创建 Cloud Storage 订阅，请运行 gcloud pubsub subscriptions create 命令。

   gcloud pubsub subscriptions create SUBSCRIPTION_ID \
       --topic=TOPIC_ID \
       --cloud-storage-bucket=BUCKET_NAME \
       --cloud-storage-file-prefix=CLOUD_STORAGE_FILE_PREFIX \
       --cloud-storage-file-suffix=CLOUD_STORAGE_FILE_SUFFIX \
       --cloud-storage-file-datetime-format=CLOUD_STORAGE_FILE_DATETIME_FORMAT \
       --cloud-storage-max-duration=CLOUD_STORAGE_MAX_DURATION \
       --cloud-storage-max-bytes=CLOUD_STORAGE_MAX_BYTES \
       --cloud-storage-max-messages=CLOUD_STORAGE_MAX_MESSAGES \
       --cloud-storage-output-format=CLOUD_STORAGE_OUTPUT_FORMAT \
       --cloud-storage-write-metadata
       --cloud-storage-use-topic-schema

   如果您想使用自定义服务账号，请将其作为附加实参提供：

   gcloud pubsub subscriptions create SUBSCRIPTION_ID \
       --topic=TOPIC_ID \
       --cloud-storage-bucket=BUCKET_NAME \
       --cloud-storage-file-prefix=CLOUD_STORAGE_FILE_PREFIX \
       --cloud-storage-file-suffix=CLOUD_STORAGE_FILE_SUFFIX \
       --cloud-storage-file-datetime-format=CLOUD_STORAGE_FILE_DATETIME_FORMAT \
       --cloud-storage-max-duration=CLOUD_STORAGE_MAX_DURATION \
       --cloud-storage-max-bytes=CLOUD_STORAGE_MAX_BYTES \
       --cloud-storage-max-messages=CLOUD_STORAGE_MAX_MESSAGES \
       --cloud-storage-output-format=CLOUD_STORAGE_OUTPUT_FORMAT \
       --cloud-storage-write-metadata
       --cloud-storage-use-topic-schema
       --cloud-storage-service-account-email=SERVICE_ACCOUNT_NAME
       

   在命令中，只有 SUBSCRIPTION_ID、--topic 标志和 --cloud-storage-bucket 标志是必需的。其余标志是可选的，可以省略。

   替换以下内容：

   • SUBSCRIPTION_ID：新 Cloud Storage 订阅的名称或 ID。
   • TOPIC_ID：主题的名称或 ID。
   • BUCKET_NAME：指定现有存储桶的名称。例如 prod_bucket。存储桶名称不得包含项目 ID。如需创建存储桶，请参阅创建存储桶。
   • CLOUD_STORAGE_FILE_PREFIX：指定 Cloud Storage 文件名的前缀。例如 log_events_。
   • CLOUD_STORAGE_FILE_SUFFIX：指定 Cloud Storage 文件名的后缀。例如 .txt。
   • CLOUD_STORAGE_FILE_DATETIME_FORMAT： 指定 Cloud Storage 文件名的日期时间格式。 例如 YYYY-MM-DD/hh_mm_ssZ。
   • CLOUD_STORAGE_MAX_DURATION：创建新的 Cloud Storage 文件之前可以经过的时长上限。该值必须介于 1m 到 10m 之间。例如 5m。
   • CLOUD_STORAGE_MAX_BYTES：在创建新文件之前可写入 Cloud Storage 文件的最大字节数。值必须介于 1KB 到 10GB 之间。例如 20MB。
   • CLOUD_STORAGE_MAX_MESSAGES：可写入 Cloud Storage 文件的消息数量上限，达到此上限后系统会创建新文件。此值必须大于或等于 1000。例如 100000。
   • CLOUD_STORAGE_OUTPUT_FORMAT：写入 Cloud Storage 的数据的输出格式。值如下所示：
     • text：消息以原始文本形式编写，并以换行符分隔。
     • avro：消息以 Avro 二进制格式写入。 --cloud-storage-write-metadata 和 --cloud-storage-use-topic-schema 仅影响输出格式为 avro 的订阅。
   • SERVICE_ACCOUNT_NAME：指定用于写入 Cloud Storage 的服务账号的名称。

C++

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

如需向 Pub/Sub 进行身份验证，请设置应用默认凭据。如需了解详情，请参阅为客户端库设置身份验证。

namespace pubsub = ::google::cloud::pubsub;
namespace pubsub_admin = ::google::cloud::pubsub_admin;
[](pubsub_admin::SubscriptionAdminClient client,
   std::string const& project_id, std::string const& topic_id,
   std::string const& subscription_id, std::string const& bucket) {
  google::pubsub::v1::Subscription request;
  request.set_name(
      pubsub::Subscription(project_id, subscription_id).FullName());
  request.set_topic(pubsub::Topic(project_id, topic_id).FullName());
  request.mutable_cloud_storage_config()->set_bucket(bucket);
  auto sub = client.CreateSubscription(request);
  if (!sub) {
    if (sub.status().code() == google::cloud::StatusCode::kAlreadyExists) {
      std::cout << "The subscription already exists\n";
      return;
    }
    throw std::move(sub).status();
  }

  std::cout << "The subscription was successfully created: "
            << sub->DebugString() << "\n";
}

C#

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

如需向 Pub/Sub 进行身份验证，请设置应用默认凭据。如需了解详情，请参阅为客户端库设置身份验证。

using Google.Cloud.PubSub.V1;
using Google.Protobuf.WellKnownTypes;
using System;

public class CreateCloudStorageSubscriptionSample
{
    public Subscription CreateCloudStorageSubscription(string projectId, string topicId, string subscriptionId,
        string bucket, string filenamePrefix, string filenameSuffix, TimeSpan maxDuration)
    {
        SubscriberServiceApiClient subscriber = SubscriberServiceApiClient.Create();
        TopicName topicName = TopicName.FromProjectTopic(projectId, topicId);
        SubscriptionName subscriptionName = SubscriptionName.FromProjectSubscription(projectId, subscriptionId);

        var subscriptionRequest = new Subscription
        {
            SubscriptionName = subscriptionName,
            TopicAsTopicName = topicName,
            CloudStorageConfig = new CloudStorageConfig
            {
                Bucket = bucket,
                FilenamePrefix = filenamePrefix,
                FilenameSuffix = filenameSuffix,
                MaxDuration = Duration.FromTimeSpan(maxDuration)
            }
        };
        var subscription = subscriber.CreateSubscription(subscriptionRequest);
        return subscription;
    }
}

Go

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

如需向 Pub/Sub 进行身份验证，请设置应用默认凭据。如需了解详情，请参阅为客户端库设置身份验证。

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

	"cloud.google.com/go/pubsub/v2"
	"cloud.google.com/go/pubsub/v2/apiv1/pubsubpb"
	"google.golang.org/protobuf/types/known/durationpb"
)

// createCloudStorageSubscription creates a Pub/Sub subscription that exports messages to Cloud Storage.
func createCloudStorageSubscription(w io.Writer, projectID, topic, subscription, bucket string) error {
	// projectID := "my-project-id"
	// topic := "projects/my-project-id/topics/my-topic"
	// subscription := "projects/my-project/subscriptions/my-sub"
	// bucket := "my-bucket" // bucket must not have the gs:// prefix
	ctx := context.Background()
	client, err := pubsub.NewClient(ctx, projectID)
	if err != nil {
		return fmt.Errorf("pubsub.NewClient: %w", err)
	}
	defer client.Close()

	sub, err := client.SubscriptionAdminClient.CreateSubscription(ctx, &pubsubpb.Subscription{
		Name:  subscription,
		Topic: topic,
		CloudStorageConfig: &pubsubpb.CloudStorageConfig{
			Bucket:         bucket,
			FilenamePrefix: "log_events_",
			FilenameSuffix: ".avro",
			OutputFormat: &pubsubpb.CloudStorageConfig_AvroConfig_{
				AvroConfig: &pubsubpb.CloudStorageConfig_AvroConfig{
					WriteMetadata: true,
				},
			},
			MaxDuration: durationpb.New(1 * time.Minute),
			MaxBytes:    1e8,
		},
	})
	if err != nil {
		return fmt.Errorf("failed to create cloud storage sub: %w", err)
	}
	fmt.Fprintf(w, "Created Cloud Storage subscription: %v\n", sub)

	return nil
}

Java

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

如需向 Pub/Sub 进行身份验证，请设置应用默认凭据。如需了解详情，请参阅为客户端库设置身份验证。

import com.google.cloud.pubsub.v1.SubscriptionAdminClient;
import com.google.protobuf.Duration;
import com.google.pubsub.v1.CloudStorageConfig;
import com.google.pubsub.v1.ProjectSubscriptionName;
import com.google.pubsub.v1.ProjectTopicName;
import com.google.pubsub.v1.Subscription;
import java.io.IOException;

public class CreateCloudStorageSubscriptionExample {
  public static void main(String... args) throws Exception {
    // TODO(developer): Replace these variables before running the sample.
    String projectId = "your-project-id";
    String topicId = "your-topic-id";
    String subscriptionId = "your-subscription-id";
    String bucket = "your-bucket";
    String filenamePrefix = "log_events_";
    String filenameSuffix = ".text";
    Duration maxDuration = Duration.newBuilder().setSeconds(300).build();

    createCloudStorageSubscription(
        projectId, topicId, subscriptionId, bucket, filenamePrefix, filenameSuffix, maxDuration);
  }

  public static void createCloudStorageSubscription(
      String projectId,
      String topicId,
      String subscriptionId,
      String bucket,
      String filenamePrefix,
      String filenameSuffix,
      Duration maxDuration)
      throws IOException {
    try (SubscriptionAdminClient subscriptionAdminClient = SubscriptionAdminClient.create()) {

      ProjectTopicName topicName = ProjectTopicName.of(projectId, topicId);
      ProjectSubscriptionName subscriptionName =
          ProjectSubscriptionName.of(projectId, subscriptionId);

      CloudStorageConfig cloudStorageConfig =
          CloudStorageConfig.newBuilder()
              .setBucket(bucket)
              .setFilenamePrefix(filenamePrefix)
              .setFilenameSuffix(filenameSuffix)
              .setMaxDuration(maxDuration)
              .build();

      Subscription subscription =
          subscriptionAdminClient.createSubscription(
              Subscription.newBuilder()
                  .setName(subscriptionName.toString())
                  .setTopic(topicName.toString())
                  .setCloudStorageConfig(cloudStorageConfig)
                  .build());

      System.out.println("Created a CloudStorage subscription: " + subscription.getAllFields());
    }
  }
}

Node.js

试用此示例之前，请按照 Pub/Sub 快速入门：使用客户端库中的 Node.js 设置说明进行操作。 如需了解详情，请参阅 Pub/Sub Node.js API 参考文档。

如需向 Pub/Sub 进行身份验证，请设置应用默认凭据。如需了解详情，请参阅为客户端库设置身份验证。

/**
 * TODO(developer): Uncomment these variables before running the sample.
 */
// const topicName = 'YOUR_TOPIC_NAME';
// const subscriptionName = 'YOUR_SUBSCRIPTION_NAME';
// const bucket = 'YOUR_BUCKET_ID';
// const filenamePrefix = 'YOUR_FILENAME_PREFIX';
// const filenameSuffix = 'YOUR_FILENAME_SUFFIX';
// const maxDuration = 60;

// Imports the Google Cloud client library
const {PubSub} = require('@google-cloud/pubsub');

// Creates a client; cache this for further use
const pubSubClient = new PubSub();

async function createCloudStorageSubscription(
  topicName,
  subscriptionName,
  bucket,
  filenamePrefix,
  filenameSuffix,
  maxDuration,
) {
  const options = {
    cloudStorageConfig: {
      bucket,
      filenamePrefix,
      filenameSuffix,
      maxDuration: {
        seconds: maxDuration,
      },
    },
  };

  await pubSubClient
    .topic(topicName)
    .createSubscription(subscriptionName, options);

  console.log(
    `Created subscription ${subscriptionName} with a cloud storage configuration.`,
  );
}

Node.js

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

如需向 Pub/Sub 进行身份验证，请设置应用默认凭据。如需了解详情，请参阅为客户端库设置身份验证。

/**
 * TODO(developer): Uncomment these variables before running the sample.
 */
// const topicName = 'YOUR_TOPIC_NAME';
// const subscriptionName = 'YOUR_SUBSCRIPTION_NAME';
// const bucket = 'YOUR_BUCKET_ID';
// const filenamePrefix = 'YOUR_FILENAME_PREFIX';
// const filenameSuffix = 'YOUR_FILENAME_SUFFIX';
// const maxDuration = 60;

// Imports the Google Cloud client library
import {CreateSubscriptionOptions, PubSub} from '@google-cloud/pubsub';

// Creates a client; cache this for further use
const pubSubClient = new PubSub();

async function createCloudStorageSubscription(
  topicName: string,
  subscriptionName: string,
  bucket: string,
  filenamePrefix: string,
  filenameSuffix: string,
  maxDuration: number,
) {
  const options: CreateSubscriptionOptions = {
    cloudStorageConfig: {
      bucket,
      filenamePrefix,
      filenameSuffix,
      maxDuration: {
        seconds: maxDuration,
      },
    },
  };

  await pubSubClient
    .topic(topicName)
    .createSubscription(subscriptionName, options);

  console.log(
    `Created subscription ${subscriptionName} with a cloud storage configuration.`,
  );
}

PHP

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

如需向 Pub/Sub 进行身份验证，请设置应用默认凭据。如需了解详情，请参阅为客户端库设置身份验证。

use Google\Cloud\PubSub\PubSubClient;

/**
 * Creates a Pub/Sub GCS subscription.
 *
 * @param string $projectId  The Google project ID.
 * @param string $topicName  The Pub/Sub topic name.
 * @param string $subscriptionName  The Pub/Sub subscription name.
 * @param string $bucket The Cloud Storage bucket name without any prefix like "gs://".
 */
function create_cloud_storage_subscription($projectId, $topicName, $subscriptionName, $bucket)
{
    $pubsub = new PubSubClient([
        'projectId' => $projectId,
    ]);
    $topic = $pubsub->topic($topicName);
    $subscription = $topic->subscription($subscriptionName);
    $config = ['bucket' => $bucket];
    $subscription->create([
        'cloudStorageConfig' => $config
    ]);

    printf('Subscription created: %s' . PHP_EOL, $subscription->name());
}

Python

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

如需向 Pub/Sub 进行身份验证，请设置应用默认凭据。如需了解详情，请参阅为客户端库设置身份验证。

from google.cloud import pubsub_v1
from google.protobuf import duration_pb2

# TODO(developer)
# project_id = "your-project-id"
# topic_id = "your-topic-id"
# subscription_id = "your-subscription-id"
# bucket = "my-bucket"

filename_prefix = "log_events_"
filename_suffix = ".avro"
# Either CloudStorageConfig.AvroConfig or CloudStorageConfig.TextConfig
# defaults to TextConfig
avro_config = pubsub_v1.types.CloudStorageConfig.AvroConfig(write_metadata=True)

publisher = pubsub_v1.PublisherClient()
subscriber = pubsub_v1.SubscriberClient()
topic_path = publisher.topic_path(project_id, topic_id)
subscription_path = subscriber.subscription_path(project_id, subscription_id)
max_duration = duration_pb2.Duration()
max_duration.FromSeconds(300)

cloudstorage_config = pubsub_v1.types.CloudStorageConfig(
    bucket=bucket,
    filename_prefix=filename_prefix,
    filename_suffix=filename_suffix,
    avro_config=avro_config,
    # Min 1 minutes, max 10 minutes
    max_duration=max_duration,
    # Min 1 KB, max 10 GiB
    max_bytes=10000000,
)

# Wrap the subscriber in a 'with' block to automatically call close() to
# close the underlying gRPC channel when done.
with subscriber:
    subscription = subscriber.create_subscription(
        request={
            "name": subscription_path,
            "topic": topic_path,
            "cloud_storage_config": cloudstorage_config,
        }
    )

print(f"CloudStorage subscription created: {subscription}.")
print(f"Bucket for subscription is: {bucket}")
print(f"Prefix is: {filename_prefix}")
print(f"Suffix is: {filename_suffix}")

监控 Cloud Storage 订阅

Cloud Monitoring 提供了许多指标来监控订阅。

如需查看与 Pub/Sub 相关的所有可用指标及其说明的列表，请参阅 Pub/Sub 的 Monitoring 文档。

您还可以在 Pub/Sub 中监控订阅。

后续步骤

• 排查 Cloud Storage 订阅问题。

• 了解 Cloud Storage。

• 查看 Pub/Sub 的价格，包括 Cloud Storage 订阅。