创建拉取订阅

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

准备工作

• 了解订阅。
• 评估您的业务是否需要拉取订阅。
• 了解拉取订阅的工作原理。

所需的角色和权限

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

此预定义角色可提供 创建拉取订阅所需的权限。如需查看所需的确切权限，请展开所需权限部分：

所需权限

创建拉取订阅需要以下权限：

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

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

跨项目订阅

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

创建拉取订阅

控制台

如需创建拉取订阅，请按以下步骤操作：

1. 在 Google Cloud 控制台中，进入订阅页面。

   前往订阅页面

2. 点击创建订阅 。

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

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

4. 在 Pub/Sub 主题 列表中，选择订阅要从中读取的主题。

5. 对于传送类型，选择拉取。

6. 可选：在转换 部分中，添加一个或多个 单条消息转换 (SMT)。如需了解详情，请参阅 创建具有 SMT 的订阅。

7. 可选：在过滤条件 字段中，输入过滤表达式以过滤订阅中的消息。如需了解详情，请参阅 过滤订阅中的消息。

8. 对于重试政策 ，选择一个选项。如需了解详情，请参阅 订阅重试政策。

9. 可选：启用死信主题以 接收无法递送的消息。

   1. 选中死信 复选框。

   2. 在死信主题 列表中，选择或创建死信主题。

   3. 在传送尝试次数上限 字段中，输入传送尝试次数上限。

10. 可选：在传送属性 部分中，启用或停用以下传送选项：

    • 仅传送一次

    • 消息排序

11. 可选：在确认期限 部分中，设置订阅者处理和确认消息的期限。如需了解详情，请参阅 使用租期管理延长确认时间。

12. 可选：在生命周期选项 部分中，配置订阅保留消息的时长。

    1. 在消息保留时长 部分中，指定未确认消息的保留时长。

    2. 如需保留已确认的消息以及未确认的消息，请选中保留已确认的消息 复选框。

    如需了解详情，请参阅 为订阅配置消息保留。

13. 可选：对于到期期限，选择一个选项：

    1. 如需设置订阅到期时间，请选中在处于非活跃状态这么多天后到期 复选框。输入订阅在被 Pub/Sub 删除之前可以保持非活跃状态的天数。

    2. 如需停用订阅到期，请选中永不过期 复选框。

    如需了解详情，请参阅 订阅到期。

14. 点击创建 。

您还可以通过主题 部分创建订阅。 此快捷方式可帮助您将主题与订阅关联。

1. 在 Google Cloud 控制台中，进入 主题 页面。

   前往主题页面

2. 点击要创建 订阅的主题旁边的 。

3. 从上下文菜单中，选择创建订阅 。

4. 在向主题添加订阅 页面中，完成上一个步骤中所述的步骤。系统会自动填写主题 ID。

gcloud

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

   激活 Cloud Shell

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

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

   gcloud pubsub subscriptions create SUBSCRIPTION_ID --topic=TOPIC_ID

   替换以下内容：

   • SUBSCRIPTION_ID：新拉取订阅的名称或 ID。
   • TOPIC_ID：主题的名称或 ID。

REST

如需创建拉取订阅，请使用 projects.subscriptions.create 方法：

请求：

必须使用 Authorization 标头中的访问令牌对请求进行身份验证。如需获取当前应用默认凭据的 访问令牌，请运行以下命令： gcloud auth application-default print-access-token。

PUT https://pubsub.googleapis.com/v1/projects/PROJECT_ID/subscriptions/SUBSCRIPTION_ID
Authorization: Bearer ACCESS_TOKEN

请求正文：

{
"topic": "projects/PROJECT_ID/topics/TOPIC_ID"
}

其中：

• PROJECT_ID 是项目 ID。
• SUBSCRIPTION_ID 是订阅 ID。
• TOPIC_ID 是主题 ID。

回答：

{
"name": "projects/PROJECT_ID/subscriptions/SUBSCRIPTION_ID",
"topic": "projects/PROJECT_ID/topics/TOPIC_ID",
"pushConfig": {},
"ackDeadlineSeconds": 10,
"messageRetentionDuration": "604800s",
"expirationPolicy": {
"ttl": "2678400s"
}
}

C++

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

namespace pubsub_admin = ::google::cloud::pubsub_admin;
namespace pubsub = ::google::cloud::pubsub;
[](pubsub_admin::SubscriptionAdminClient client,
   std::string const& project_id, std::string const& topic_id,
   std::string const& subscription_id) {
  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());
  auto sub = client.CreateSubscription(request);
  if (sub.status().code() == google::cloud::StatusCode::kAlreadyExists) {
    std::cout << "The subscription already exists\n";
    return;
  }
  if (!sub) throw std::move(sub).status();

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

C#

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

using Google.Cloud.PubSub.V1;
using Grpc.Core;

public class CreateSubscriptionSample
{
    public Subscription CreateSubscription(string projectId, string topicId, string subscriptionId)
    {
        SubscriberServiceApiClient subscriber = SubscriberServiceApiClient.Create();
        TopicName topicName = TopicName.FromProjectTopic(projectId, topicId);

        SubscriptionName subscriptionName = SubscriptionName.FromProjectSubscription(projectId, subscriptionId);
        Subscription subscription = null;

        try
        {
            subscription = subscriber.CreateSubscription(subscriptionName, topicName, pushConfig: null, ackDeadlineSeconds: 60);
        }
        catch (RpcException e) when (e.Status.StatusCode == StatusCode.AlreadyExists)
        {
            // Already exists.  That's fine.
        }
        return subscription;
    }
}

Go

以下示例使用 Go Pub/Sub 客户端库的主要版本 (v2)。如果您仍在使用 v1 库，请参阅 迁移到 v2 的指南。 如需查看 v1 代码示例列表，请参阅 已废弃的代码示例。

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

import (
	"context"
	"fmt"
	"io"

	"cloud.google.com/go/pubsub/v2"
	"cloud.google.com/go/pubsub/v2/apiv1/pubsubpb"
)

func create(w io.Writer, projectID, topic, subscription string) error {
	// projectID := "my-project-id"
	// topic := "projects/my-project-id/topics/my-topic"
	// subscription := "projects/my-project/subscriptions/my-sub"
	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,
	})
	if err != nil {
		return fmt.Errorf("CreateSubscription: %w", err)
	}
	fmt.Fprintf(w, "Created subscription: %v\n", sub)
	return nil
}

Java

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

import com.google.cloud.pubsub.v1.SubscriptionAdminClient;
import com.google.pubsub.v1.PushConfig;
import com.google.pubsub.v1.Subscription;
import com.google.pubsub.v1.SubscriptionName;
import com.google.pubsub.v1.TopicName;
import java.io.IOException;

public class CreatePullSubscriptionExample {
  public static void main(String... args) throws Exception {
    // TODO(developer): Replace these variables before running the sample.
    String projectId = "your-project-id";
    String subscriptionId = "your-subscription-id";
    String topicId = "your-topic-id";

    createPullSubscriptionExample(projectId, subscriptionId, topicId);
  }

  public static void createPullSubscriptionExample(
      String projectId, String subscriptionId, String topicId) throws IOException {
    try (SubscriptionAdminClient subscriptionAdminClient = SubscriptionAdminClient.create()) {
      TopicName topicName = TopicName.of(projectId, topicId);
      SubscriptionName subscriptionName = SubscriptionName.of(projectId, subscriptionId);
      // Create a pull subscription with default acknowledgement deadline of 10 seconds.
      // Messages not successfully acknowledged within 10 seconds will get resent by the server.
      Subscription subscription =
          subscriptionAdminClient.createSubscription(
              subscriptionName, topicName, PushConfig.getDefaultInstance(), 10);
      System.out.println("Created pull subscription: " + subscription.getName());
    }
  }
}

Node.js

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

/**
 * TODO(developer): Uncomment these variables before running the sample.
 */
// const topicNameOrId = 'YOUR_TOPIC_NAME_OR_ID';
// const subscriptionNameOrId = 'YOUR_SUBSCRIPTION_NAME_OR_ID';

// 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 createSubscription(topicNameOrId, subscriptionNameOrId) {
  // Creates a new subscription
  await pubSubClient
    .topic(topicNameOrId)
    .createSubscription(subscriptionNameOrId);
  console.log(`Subscription ${subscriptionNameOrId} created.`);
}

Node.ts

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

/**
 * TODO(developer): Uncomment these variables before running the sample.
 */
// const topicNameOrId = 'YOUR_TOPIC_NAME_OR_ID';
// const subscriptionNameOrId = 'YOUR_SUBSCRIPTION_NAME_OR_ID';

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

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

async function createSubscription(
  topicNameOrId: string,
  subscriptionNameOrId: string,
) {
  // Creates a new subscription
  await pubSubClient
    .topic(topicNameOrId)
    .createSubscription(subscriptionNameOrId);
  console.log(`Subscription ${subscriptionNameOrId} created.`);
}

PHP

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

use Google\Cloud\PubSub\PubSubClient;

/**
 * Creates a Pub/Sub subscription.
 *
 * @param string $projectId  The Google project ID.
 * @param string $topicName  The Pub/Sub topic name.
 * @param string $subscriptionName  The Pub/Sub subscription name.
 */
function create_subscription($projectId, $topicName, $subscriptionName)
{
    $pubsub = new PubSubClient([
        'projectId' => $projectId,
    ]);
    $topic = $pubsub->topic($topicName);
    $subscription = $topic->subscription($subscriptionName);
    $subscription->create();

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

Python

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

from google.cloud import pubsub_v1

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

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)

# 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}
    )

print(f"Subscription created: {subscription}")

Ruby

以下示例使用 Ruby Pub/Sub 客户端库 v3。如果您仍在使用 v2 库，请参阅 迁移到 v3 的指南。 如需查看 Ruby v2 代码示例列表，请参阅 已废弃的代码示例。

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

# topic_id        = "your-topic-id"
# subscription_id = "your-subscription-id"

pubsub = Google::Cloud::PubSub.new
subscription_admin = pubsub.subscription_admin

subscription = subscription_admin.create_subscription \
  name: pubsub.subscription_path(subscription_id),
  topic: pubsub.topic_path(topic_id)

puts "Pull subscription #{subscription_id} created."

监控拉取订阅

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

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

您还可以从 Pub/Sub 内 监控订阅。

后续步骤

• 接收来自拉取订阅的消息。
• 使用 gcloud 命令创建或修改订阅。
• 使用 REST API 创建或修改订阅。