创建 Cloud Storage 订阅

存储分区权限

请执行以下步骤，在存储桶级别授予 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 服务账号的名称。

   服务账号的格式为 service-PROJECT_NUMBER@gcp-sa-pubsub.iam.gserviceaccount.com。例如，对于 PROJECT_NUMBER 为 112233445566 的项目，服务账号的格式为 service-112233445566@gcp-sa-pubsub.iam.gserviceaccount.com。

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 服务账号的名称。

   服务账号的格式为 service-PROJECT_NUMBER@gcp-sa-pubsub.iam.gserviceaccount.com。例如，对于 PROJECT_NUMBER 为 112233445566 的项目，服务账号的格式为 service-112233445566@gcp-sa-pubsub.iam.gserviceaccount.com。

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

5. 点击保存。

控制台

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

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

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 文件的字节数上限。值必须介于 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 的订阅。

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