创建 Cloud Storage 订阅

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

准备工作

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

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

所需的角色和权限

以下是有关角色和权限的一系列指南：

• 如需创建订阅，您必须在项目级别配置访问权限控制。

• 如果您的订阅和主题位于不同的项目中，您还需要拥有资源级权限，如本部分后面所述。

• 要创建 Cloud Storage 订阅， Pub/Sub 服务账号必须具有写入 特定 Cloud Storage 存储桶以及读取存储桶元数据。如需详细了解如何授予这些权限，请参阅本文档的下一部分。

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

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

所需权限

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

• 创建订阅： pubsub.subscriptions.create
• 将订阅附加到主题： pubsub.topics.attachSubscription
• 从订阅中拉取： pubsub.subscriptions.consume
• 获取订阅： pubsub.subscriptions.get
• 列出订阅： pubsub.subscriptions.list
• 更新订阅： pubsub.subscriptions.update
• 删除订阅： pubsub.subscriptions.delete
• 获取订阅的 IAM 政策： pubsub.subscriptions.getIamPolicy
• 为订阅配置 IAM 政策： pubsub.subscriptions.setIamPolicy

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

如果您需要创建 Cloud Storage 一个项目中的订阅与另一个项目中的主题关联 项目，请让主题管理员也授予您 Pub/Sub Editor (roles/pubsub.editor) IAM 角色。

为 Pub/Sub 服务账号分配 Cloud Storage 角色

某些 Google Cloud 服务具有由 Google Cloud 代管式服务账号， 可让服务访问您的资源这些服务账号称为 服务代理Pub/Sub 会为每个项目创建并维护一个服务账号，格式为 [e-mail]。

如需创建 Cloud Storage 订阅，Pub/Sub 服务账号必须有权写入特定 Cloud Storage 存储桶并读取存储桶元数据。选择以下任一过程：

• 在存储桶级别授予权限。在特定 Cloud Storage 中 授予 Storage Object Creator (roles/storage.objectCreator) 角色 和 Storage Legacy Bucket Reader (roles/storage.legacyBucketReader) 角色 Pub/Sub 服务账号。

• 如果您必须在项目级授予角色，则可以改为在包含 Cloud Storage 存储桶的项目中授予 Storage Admin (roles/storage.admin) 角色。将此角色授予 Pub/Sub 服务账号。

存储分区权限

如需授予 Storage Object Creator，请执行以下步骤 (roles/storage.objectCreator) 和 Storage Legacy Bucket Reader (roles/storage.legacyBucketReader) 存储桶级角色：

1. 在 Google Cloud 控制台中，转到 Cloud Storage 页面。

   转到 Cloud Storage

2. 点击您要将消息写入到的 Cloud Storage 存储桶。

   系统随即会打开存储分区详情页面。

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

4. 在权限 > 按主账号查看标签页中，点击授予访问权限。

   系统随即会打开授予访问权限页面。

5. 在添加主账号部分中，输入 Pub/Sub 的名称。 服务账号。

   服务账号的格式为 [e-mail]。 例如，对于 PROJECT_NUMBER 为 112233445566 的项目，服务账号的格式为 [e-mail]。

6. 在分配角色 > 选择角色下拉菜单中，输入 Creator，然后选择 Storage Object Creator 角色。

7. 点击添加其他角色。

8. 在选择角色下拉菜单中，输入 Bucket Reader，然后选择 Storage Legacy Bucket Reader 角色。

9. 点击保存。

项目权限

如需授予 Storage Admin，请执行以下步骤 (roles/storage.admin) 角色：

1. 在 Google Cloud 控制台中，转到 IAM 页面。

   转到 IAM

2. 在权限 > 按主账号查看标签页中，点击授予访问权限。

   系统随即会打开授予访问权限页面。

3. 在添加主账号部分中，输入 Pub/Sub 的名称。 服务账号。

   服务账号的格式为 [e-mail]。 例如，对于 PROJECT_NUMBER=112233445566 的项目， 服务账号的格式为 [e-mail]。

4. 在分配角色 > 选择角色下拉菜单中，输入 Storage Admin，然后选择 Storage Admin 角色。

5. 点击保存。

如需详细了解 Cloud Storage IAM，请参阅 Cloud Storage Identity and Access Management。

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 批处理条件：

• 批量存储时长上限。此设置是必需的。通过 如果出现以下情况，Cloud Storage 订阅会写入新的输出文件： 超过指定的时长上限值。如果您 指定值，则系统会应用默认值 5 分钟。 以下是适用于最长时长的值：

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

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

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

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

  • 最小值 = 1000

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

Cloud Storage 订阅可能会 Cloud Storage 存储桶。如果您已将订阅配置为每 6 分钟创建一个新文件，则可能会发现系统每 6 分钟就会创建多个 Cloud Storage 文件。

在某些情况下，Pub/Sub 可能会开始 文件。 如果进行订阅，文件也可能会超过“最大字节数”值 接收大于最大字节数值的消息。

文件格式

创建 Cloud Storage 订阅时，您可以将要存储在 Cloud Storage 存储桶中的输出文件的格式指定为文本或 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 将展开记录的字段列表以包含元数据字段。因此，记录不得包含与元数据字段（subscription_name、message_id、publish_time 或 attributes）同名的任何字段。

创建 Cloud Storage 订阅

控制台

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

   前往“订阅”页面

2. 点击创建订阅。

3. 对于订阅 ID 字段，输入名称。

   要了解如何命名订阅，请参阅主题命名准则或 订阅。

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

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

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

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

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

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

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

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

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

   指定存储桶时，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. 我们强烈建议您启用死信来处理邮件传送失败问题。

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

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

gcloud

1. In the Google Cloud console, activate Cloud Shell.

   Activate Cloud Shell

   At the bottom of the Google Cloud console, a Cloud Shell session starts and displays a command-line prompt. Cloud Shell is a shell environment with the Google Cloud CLI already installed and with values already set for your current project. It can take a few seconds for the session to initialize.

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

   在此命令中，只有 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 文件 创建。该值必须介于 1 米到 10 米之间。例如 5m。
   • CLOUD_STORAGE_MAX_BYTES：在创建新文件之前，可以写入 Cloud Storage 文件的最大字节数。该值必须介于 1 KB 到 10 GB 之间。例如 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 的订阅。

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

Go

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

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

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

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

// createCloudStorageSubscription creates a Pub/Sub subscription that exports messages to Cloud Storage.
func createCloudStorageSubscription(w io.Writer, projectID, subID string, topic *pubsub.Topic, bucket string) error {
	// projectID := "my-project-id"
	// subID := "my-sub"
	// topic of type https://godoc.org/cloud.google.com/go/pubsub#Topic
	// note bucket should not have the gs:// prefix
	// bucket := "my-bucket"
	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.CreateSubscription(ctx, subID, pubsub.SubscriptionConfig{
		Topic: topic,
		CloudStorageConfig: pubsub.CloudStorageConfig{
			Bucket:         bucket,
			FilenamePrefix: "log_events_",
			FilenameSuffix: ".avro",
			OutputFormat:   &pubsub.CloudStorageOutputFormatAvroConfig{WriteMetadata: true},
			MaxDuration:    1 * time.Minute,
			MaxBytes:       1e8,
		},
	})
	if err != nil {
		return fmt.Errorf("client.CreateSubscription: %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());
    }
  }
}

监控 Cloud Storage 订阅

Cloud Monitoring 提供了一些指标来监控订阅。

获取与 Pub/Sub 相关的所有可用指标的列表 及其说明，请参阅 Pub/Sub 的监控文档。

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

后续步骤

• 排查 Cloud Storage 订阅问题。

• 了解 Cloud Storage。

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