创建拉取订阅

本文档介绍了如何创建拉取订阅。您可以使用 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. 选中 Dead lettering 复选框。

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

Rust

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

use google_cloud_pubsub::client::SubscriptionAdmin;

pub async fn sample(
    client: &SubscriptionAdmin,
    project_id: &str,
    topic_id: &str,
    subscription_id: &str,
) -> anyhow::Result<()> {
    let subscription_name = format!("projects/{project_id}/subscriptions/{subscription_id}");
    let topic_name = format!("projects/{project_id}/topics/{topic_id}");

    let subscription = client
        .create_subscription()
        .set_name(subscription_name)
        .set_topic(topic_name)
        .send()
        .await?;

    println!("successfully created subscription {subscription:?}");
    Ok(())
}

监控拉取订阅

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

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

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

后续步骤

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