将 Pub/Sub 精简版消息导出到 Pub/Sub

本文档介绍如何设置将 Pub/Sub Lite 消息自动导出到 Pub/Sub 的设置。

在下列情况下，您可能会使用此功能：

  • 在混合使用 Pub/Sub Lite 和 Pub/Sub 的工作负载之间互操作。
  • 将 Pub/Sub Lite 工作负载迁移到 Pub/Sub。
  • 使用基于 Pub/Sub Lite 的现有应用中的 Pub/Sub 高级功能，例如推送订阅和过滤。
  • 整合多个数据流水线。

概览

如需将消息从 Pub/Sub Lite 导出到 Pub/Sub，您需要创建一个特殊类型的订阅，称为“导出订阅”。导出订阅接收来自精简版主题的消息，将其转换为 Pub/Sub 消息，并将转换后的消息发送到目标 Pub/Sub 主题。

导出 Pub/Sub Lite 消息的图表

精简版主题可以同时包含导出订阅和标准订阅。这两种订阅类型在配额用量和预留吞吐量方面相同。导出订阅会使用精简版订阅吞吐量容量，并按 Pub/Sub 发布吞吐量收费。

导出订阅会将一个精简版主题恰好连接到一个 Pub/Sub 主题。但是，一个精简版主题可以有多个连接到不同 Pub/Sub 主题的导出订阅（扇出架构）。您还可以从多个精简版主题导出到同一 Pub/Sub 主题（扇入架构）。

Authentication

导出订阅会访问 Pub/Sub Lite 和 Pub/Sub 资源。如需创建导出订阅，您需要以下权限：

  • pubsublite.subscriptions.create。以下预定义角色包含此权限：

    • roles/pubsublite.admin
    • roles/pubsublite.editor

    请参阅 Pub/Sub Lite 的访问权限控制。

  • pubsub.topics.get。以下预定义角色包含此权限：

    • roles/pubsub.admin
    • roles/pubsub.editor
    • roles/pubsub.viewer

    请参阅 Pub/Sub 的访问权限控制。

服务代理

导出订阅会代表您发布到 Pub/Sub 主题。为此，它使用 Google 管理的服务帐号。

在项目中创建第一个导出订阅时，系统会自动创建由 Google 管理的 Pub/Sub Lite 服务代理。如果您在同一项目中创建其他导出订阅，这些订阅将使用相同的服务代理。服务代理采用以下命名方案：service-<your_project_number>@gcp-sa-pubsublite.iam.gserviceaccount.com。

创建的服务代理具有发布到导出订阅所在项目内的所有 Pub/Sub 和 Pub/Sub Lite 主题的权限。如果目标 Pub/Sub 主题与导出订阅位于不同的项目中，您必须添加 Pub/Sub Publisher 角色 (roles/pubsub.publisher) 来向服务代理授予额外的权限。您可以为整个项目或单个主题授予权限。我们建议遵循最小权限原则在主题级别授予权限。

如需了解详情，请参阅通过 Google Cloud 控制台控制访问权限。您还可以使用 gcloud projects add-iam-policy-binding 命令添加 IAM 角色：

gcloud pubsub topics add-iam-policy-binding TOPIC_NAME \
 --member=serviceAccount:service-PROJECT_NUMBER@gcp-sa-pubsublite.iam.gserviceaccount.com
 --role=roles/pubsub.publisher

替换以下内容：

  • TOPIC_NAME：要添加 IAM 政策绑定的目标 Pub/Sub 主题的名称。
  • PROJECT_NUMBER：Pub/Sub Lite 导出订阅的项目的项目编号。

创建导出订阅

您可以使用 Google Cloud 控制台、Google Cloud CLI 或 Pub/Sub Lite API 创建精简版导出订阅。

精简版导出订阅必须与其所属的精简版主题位于同一项目和位置中。如需创建精简版主题，请参阅创建和管理精简版主题。

如果您将导出订阅附加到精简版主题，请确保发布到精简版主题的所有消息都与 Pub/Sub 兼容。如需了解详情，请参阅邮件兼容性。

创建导出订阅后，您便无法将导出订阅更改为标准订阅，反之亦然。

控制台

  1. 前往精简版订阅页面。

    前往“精简版订阅”

  2. 点击创建精简版订阅。

  3. 输入精简版订阅 ID。

  4. 选择从其接收消息的精简版主题。

  5. 选择立即传送消息或存储后传送消息。

  6. 选择一种起始偏移量。

  7. 选择导出到 Pub/Sub 主题。

  8. 在目标主题列表中，选择一个 Pub/Sub 主题来接收导出的精简版消息。

  9. 可选。指定死信主题。

    1. 选中启用死信复选框。
    2. 选择一个精简版主题作为死信主题，或点击创建精简版主题以创建新的死信主题。死信主题必须与导出订阅位于同一位置（可用区或区域）和项目。

  10. 点击创建。

gcloud

如需创建导出订阅，请使用 gcloud pubsub lite-subscriptions create 命令：

gcloud pubsub lite-subscriptions create SUBSCRIPTION_ID \
  --location=LOCATION \
  --topic=TOPIC_ID \
  --export-pubsub-topic=PUBSUB_TOPIC_NAME \
  --export-dead-letter-topic=DEAD_LETTER_TOPIC_ID \
  --export-desired-state=DESIRED_STATE

替换以下内容：

  • SUBSCRIPTION_ID：要创建的精简版订阅的 ID。
  • LOCATION：精简版订阅的位置。
  • TOPIC_ID：要附加到精简版订阅的精简版主题的 ID。
  • PUBSUB_TOPIC_NAME：要导出到的 Pub/Sub 主题的名称。如果主题属于其他项目，请指定全名：projects/my-project-id/topics/my-topic-id。
  • DEAD_LETTER_TOPIC_ID：可选。要用作死信主题的精简版主题的 ID。死信主题必须与导出订阅位于同一位置（可用区或区域）和项目。
  • DESIRED_STATE：可选。订阅的起始状态。 支持使用以下值：
    • active：订阅会将精简版消息导出到 Pub/Sub。（默认）。
    • paused：精简版消息导出已暂停。

如果请求成功，命令行会显示一条确认消息：

Created [SUBSCRIPTION_ID].

协议

如需创建精简版导出订阅，请发送如下所示的 POST 请求：

POST https://REGION-pubsublite.googleapis.com/v1/admin/projects/PROJECT_NUMBER/locations/LOCATION/subscriptions/SUBSCRIPTION_ID
Authorization: Bearer $(gcloud auth print-access-token)

替换以下内容：

  • REGION：用于存储精简版订阅的区域。
  • PROJECT_NUMBER：要在其中创建精简版订阅的项目的编号。
  • LOCATION：Pub/Sub Lite 支持的位置的名称。
  • SUBSCRIPTION_ID：精简版订阅的 ID。

在请求正文中指定以下字段：

{
  "topic": "projects/PROJECT_NUMBER/locations/LOCATION/topics/TOPIC_ID",
  "deliveryConfig": {
      "deliveryRequirement": "DELIVERY_REQUIREMENT",
  },
  "exportConfig": {
      "desiredState": "DESIRED_STATE",
      "deadLetterTopic": "projects/PROJECT_NUMBER/locations/LOCATION/topics/DEAD_LETTER_TOPIC_ID",
      "pubsubConfig": {
          "topic": "PUBSUB_TOPIC_NAME"
      }
  }
}

替换以下内容：

  • DELIVERY_REQUIREMENT：传送要求，可为 DELIVER_AFTER_STORED 或 DELIVER_IMMEDIATELY。
  • DESIRED_STATE：订阅的起始状态。支持以下值：
    • ACTIVE：订阅会将精简版消息导出到 Pub/Sub。
    • PAUSED：精简版消息导出已暂停。
  • DEAD_LETTER_TOPIC_ID：要用作死信主题的现有精简版主题的 ID。该主题必须与导出订阅本身位于同一位置（可用区或区域）和项目。
  • PUBSUB_TOPIC_NAME：要导出到的 Pub/Sub 主题的名称。示例格式：projects/my-project-id/topics/my-topic-id。

如果请求成功，则响应是 JSON 格式的精简版订阅：

{
  "deliveryConfig": {
      "deliveryRequirement": "DELIVERY_REQUIREMENT",
  },
  "exportConfig": {
      "desiredState": "DESIRED_STATE",
      "deadLetterTopic": "projects/PROJECT_NUMBER/locations/LOCATION/topics/DEAD_LETTER_TOPIC_ID",
      "pubsubConfig": {
          "topic": "PUBSUB_TOPIC_NAME"
      },
  "name": "projects/PROJECT_NUMBER/locations/LOCATION/subscriptions/SUBSCRIPTION_ID",
  "topic": "projects/PROJECT_NUMBER/locations/LOCATION/topics/TOPIC_ID",
}

Go

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

import (
	"context"
	"fmt"
	"io"

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

func createPubsubExportSubscription(w io.Writer, projectID, region, location, topicID, subID, pubsubTopicID string) error {
	// projectID := "my-project-id"
	// region := "us-central1"
	// NOTE: location can be either a region ("us-central1") or a zone ("us-central1-a")
	// For a list of valid locations, see https://cloud.google.com/pubsub/lite/docs/locations.
	// location := "us-central1"
	// NOTE: topic and subscription must be in the same region/zone (e.g. "us-central1-a")
	// topicID := "my-topic"
	// subID := "my-subscription"
	// pubsubTopicID := "destination-topic-id"
	ctx := context.Background()
	client, err := pubsublite.NewAdminClient(ctx, region)
	if err != nil {
		return fmt.Errorf("pubsublite.NewAdminClient: %w", err)
	}
	defer client.Close()

	// Initialize the subscription to the oldest retained messages for each
	// partition.
	targetLocation := pubsublite.AtTargetLocation(pubsublite.Beginning)

	sub, err := client.CreateSubscription(ctx, pubsublite.SubscriptionConfig{
		Name:                fmt.Sprintf("projects/%s/locations/%s/subscriptions/%s", projectID, location, subID),
		Topic:               fmt.Sprintf("projects/%s/locations/%s/topics/%s", projectID, location, topicID),
		DeliveryRequirement: pubsublite.DeliverImmediately, // Can also be DeliverAfterStored.
		// Configures an export subscription that writes messages to a Pub/Sub topic.
		ExportConfig: &pubsublite.ExportConfig{
			DesiredState: pubsublite.ExportActive, // Can also be ExportPaused.
			Destination: &pubsublite.PubSubDestinationConfig{
				Topic: fmt.Sprintf("projects/%s/topics/%s", projectID, pubsubTopicID),
			},
		},
	}, targetLocation)
	if err != nil {
		return fmt.Errorf("client.CreateSubscription got err: %w", err)
	}
	fmt.Fprintf(w, "Created export subscription: %s\n", sub.Name)
	return nil
}

Java

在运行此示例之前，请按照 Pub/Sub Lite 客户端库中的 Java 设置说明进行操作。

import com.google.api.gax.rpc.AlreadyExistsException;
import com.google.cloud.pubsublite.AdminClient;
import com.google.cloud.pubsublite.AdminClientSettings;
import com.google.cloud.pubsublite.BacklogLocation;
import com.google.cloud.pubsublite.CloudRegion;
import com.google.cloud.pubsublite.CloudRegionOrZone;
import com.google.cloud.pubsublite.CloudZone;
import com.google.cloud.pubsublite.ProjectNumber;
import com.google.cloud.pubsublite.SeekTarget;
import com.google.cloud.pubsublite.SubscriptionName;
import com.google.cloud.pubsublite.SubscriptionPath;
import com.google.cloud.pubsublite.TopicName;
import com.google.cloud.pubsublite.TopicPath;
import com.google.cloud.pubsublite.proto.ExportConfig;
import com.google.cloud.pubsublite.proto.ExportConfig.PubSubConfig;
import com.google.cloud.pubsublite.proto.ExportConfig.State;
import com.google.cloud.pubsublite.proto.Subscription;
import com.google.cloud.pubsublite.proto.Subscription.DeliveryConfig;
import com.google.cloud.pubsublite.proto.Subscription.DeliveryConfig.DeliveryRequirement;
import java.util.concurrent.ExecutionException;

public class CreatePubsubExportSubscriptionExample {

  public static void main(String... args) throws Exception {
    // TODO(developer): Replace these variables before running the sample.
    String cloudRegion = "your-cloud-region";
    char zoneId = 'b';
    String topicId = "your-topic-id";
    String subscriptionId = "your-subscription-id";
    String pubsubTopicId = "destination-topic-id";
    long projectNumber = Long.parseLong("123456789");
    // True if using a regional location. False if using a zonal location.
    // https://cloud.google.com/pubsub/lite/docs/topics
    boolean regional = false;

    createPubsubExportSubscriptionExample(
        cloudRegion, zoneId, projectNumber, topicId, subscriptionId, pubsubTopicId, regional);
  }

  public static void createPubsubExportSubscriptionExample(
      String cloudRegion,
      char zoneId,
      long projectNumber,
      String topicId,
      String subscriptionId,
      String pubsubTopicId,
      boolean regional)
      throws Exception {

    CloudRegionOrZone location;
    if (regional) {
      location = CloudRegionOrZone.of(CloudRegion.of(cloudRegion));
    } else {
      location = CloudRegionOrZone.of(CloudZone.of(CloudRegion.of(cloudRegion), zoneId));
    }

    TopicPath topicPath =
        TopicPath.newBuilder()
            .setProject(ProjectNumber.of(projectNumber))
            .setLocation(location)
            .setName(TopicName.of(topicId))
            .build();

    SubscriptionPath subscriptionPath =
        SubscriptionPath.newBuilder()
            .setLocation(location)
            .setProject(ProjectNumber.of(projectNumber))
            .setName(SubscriptionName.of(subscriptionId))
            .build();

    com.google.pubsub.v1.TopicName pubsubTopicName =
        com.google.pubsub.v1.TopicName.of(String.valueOf(projectNumber), pubsubTopicId);

    Subscription subscription =
        Subscription.newBuilder()
            .setDeliveryConfig(
                // Possible values for DeliveryRequirement:
                // - `DELIVER_IMMEDIATELY`
                // - `DELIVER_AFTER_STORED`
                // You may choose whether to wait for a published message to be successfully written
                // to storage before the server delivers it to subscribers. `DELIVER_IMMEDIATELY` is
                // suitable for applications that need higher throughput.
                DeliveryConfig.newBuilder()
                    .setDeliveryRequirement(DeliveryRequirement.DELIVER_IMMEDIATELY))
            .setExportConfig(
                // Configures an export subscription that writes messages to a Pub/Sub topic.
                ExportConfig.newBuilder()
                    // Possible values for State:
                    // - `ACTIVE`: enable message processing.
                    // - `PAUSED`: suspend message processing.
                    .setDesiredState(State.ACTIVE)
                    .setPubsubConfig(
                        PubSubConfig.newBuilder().setTopic(pubsubTopicName.toString())))
            .setName(subscriptionPath.toString())
            .setTopic(topicPath.toString())
            .build();

    // Target location within the message backlog that the subscription should be initialized to.
    SeekTarget initialLocation = SeekTarget.of(BacklogLocation.BEGINNING);

    AdminClientSettings adminClientSettings =
        AdminClientSettings.newBuilder().setRegion(location.extractRegion()).build();

    // Initialize client that will be used to send requests. This client only needs to be created
    // once, and can be reused for multiple requests. After completing all of your requests, call
    // the "close" method on the client to safely clean up any remaining background resources, or
    // use "try-with-close" statement to do this automatically.
    try (AdminClient adminClient = AdminClient.create(adminClientSettings)) {
      Subscription response = adminClient.createSubscription(subscription, initialLocation).get();
      System.out.println(response.getAllFields() + " created successfully.");
    } catch (ExecutionException e) {
      try {
        throw e.getCause();
      } catch (AlreadyExistsException alreadyExists) {
        System.out.println("This subscription already exists.");
      } catch (Throwable throwable) {
        throwable.printStackTrace();
      }
    }
  }
}

Python

在运行此示例之前，请按照 Pub/Sub Lite 客户端库中的 Python 设置说明进行操作。

from google.api_core.exceptions import AlreadyExists
from google.cloud.pubsub_v1 import PublisherClient
from google.cloud.pubsublite import AdminClient, Subscription, ExportConfig
from google.cloud.pubsublite.types import (
    BacklogLocation,
    CloudRegion,
    CloudZone,
    SubscriptionPath,
    TopicPath,
)

def create_lite_pubsub_export_subscription(
    project_number,
    cloud_region="us-central1",
    zone_id="a",
    topic_id="my-topic-id",
    subscription_id="my-subscription-id",
    pubsub_topic_id="destination-topic-id",
    regional=True,
    target_location=BacklogLocation.BEGINNING,
):
    if regional:
        location = CloudRegion(cloud_region)
    else:
        location = CloudZone(CloudRegion(cloud_region), zone_id)

    topic_path = TopicPath(project_number, location, topic_id)
    subscription_path = SubscriptionPath(project_number, location, subscription_id)
    destination_topic_path = PublisherClient.topic_path(project_number, pubsub_topic_id)

    subscription = Subscription(
        name=str(subscription_path),
        topic=str(topic_path),
        delivery_config=Subscription.DeliveryConfig(
            # Possible values for delivery_requirement:
            # - `DELIVER_IMMEDIATELY`
            # - `DELIVER_AFTER_STORED`
            # You may choose whether to wait for a published message to be successfully written
            # to storage before the server delivers it to subscribers. `DELIVER_IMMEDIATELY` is
            # suitable for applications that need higher throughput.
            delivery_requirement=Subscription.DeliveryConfig.DeliveryRequirement.DELIVER_IMMEDIATELY,
        ),
        # Configures an export subscription that writes messages to a Pub/Sub topic.
        export_config=ExportConfig(
            # Possible values for desired_state:
            # - `ACTIVE`: enable message processing.
            # - `PAUSED`: suspend message processing.
            desired_state=ExportConfig.State.ACTIVE,
            pubsub_config=ExportConfig.PubSubConfig(
                topic=destination_topic_path,
            ),
        ),
    )

    # Initialize client that will be used to send requests across threads. This
    # client only needs to be created once, and can be reused for multiple requests.
    client = AdminClient(cloud_region)
    try:
        response = client.create_subscription(subscription, target_location)
        print(f"{response.name} created successfully.")
    except AlreadyExists:
        print(f"{subscription_path} already exists.")

更新导出订阅

您可以使用 Google Cloud 控制台、Google Cloud CLI 或 Pub/Sub Lite API 更新精简版订阅。新设置最多可能需要 30 秒才能应用。

控制台

  1. 前往精简版订阅页面。

    前往“精简版订阅”

  2. 点击精简版订阅 ID。

  3. 在精简版订阅详情页面中，点击修改。

gCloud

要更新精简版订阅，请使用 gcloud pubsub lite-subscriptions update 命令：

gcloud pubsub lite-subscriptions update SUBSCRIPTION_ID \
--location=LOCATION \
--delivery-requirement=DELIVERY_REQUIREMENT \
--export-pubsub-topic=PUBSUB_TOPIC_NAME \
--export-dead-letter-topic=DEAD_LETTER_TOPIC_ID \
--export-desired-state=DESIRED_STATE

替换以下内容：

  • SUBSCRIPTION_ID：精简版订阅的 ID
  • LOCATION：精简版订阅的位置。
  • DELIVERY_REQUIREMENT：可选。传送要求，值为 deliver-after-stored 或 deliver-immediately。
  • PUBSUB_TOPIC_NAME：可选。要导出到的 Pub/Sub 主题的名称。如果主题位于其他项目中，请指定全名：projects/my-project-id/topics/my-topic-id。
  • DEAD_LETTER_TOPIC_ID：要用作死信主题的现有精简版主题的 ID。该主题必须与导出订阅本身位于同一位置（可用区或区域）和项目。
  • DESIRED_STATE：可选。所需的订阅状态。 支持使用以下值：
    • active：订阅会将精简版消息导出到 Pub/Sub。（默认）。
    • paused：精简版消息导出已暂停。

如果请求成功，命令行将显示精简版订阅：

Updated subscription [SUBSCRIPTION_ID].
deliveryConfig:
deliveryRequirement: DELIVERY_REQUIREMENT
exportConfig:
currentState: DESIRED_STATE
deadLetterTopic: projects/PROJECT_NUMBER/locations/LOCATION/topics/DEAD_LETTER_TOPIC_ID
desiredState: DESIRED_STATE
pubsubConfig:
  topic: PUBSUB_TOPIC_NAME
name: projects/PROJECT_NUMBER/locations/LOCATION/subscriptions/SUBSCRIPTION_ID
topic: projects/PROJECT_NUMBER/locations/LOCATION/topics/TOPIC_ID

协议

要更新精简版订阅，请发送 PATCH 请求，如下所示：

PATCH https://REGION-pubsublite.googleapis.com/v1/admin/projects/PROJECT_NUMBER/locations/LOCATION/subscriptions/SUBSCRIPTION_ID?updateMask=deliveryConfig.deliveryRequirement,exportConfig
Authorization: Bearer $(gcloud auth print-access-token)

替换以下内容：

  • REGION：创建精简版订阅的区域。
  • PROJECT_NUMBER：创建了精简版订阅的项目的编号。
  • LOCATION：创建精简版订阅的位置。
  • SUBSCRIPTION_ID：精简版订阅的 ID。

在请求正文中指定以下字段：

{
  "deliveryConfig": {
      "deliveryRequirement": "DELIVERY_REQUIREMENT",
  },
  "exportConfig": {
      "desiredState": "DESIRED_STATE",
      "deadLetterTopic": "projects/PROJECT_NUMBER/locations/LOCATION/topics/DEAD_LETTER_TOPIC_ID",
      "pubsubConfig": {
          "topic": "PUBSUB_TOPIC_NAME"
      }
  }
}

替换以下内容：

  • DELIVERY_REQUIREMENT：传送要求，可为 DELIVER_AFTER_STORED 或 DELIVER_IMMEDIATELY。
  • DESIRED_STATE：期望的订阅状态。支持以下值：
    • ACTIVE：订阅会将精简版消息导出到 Pub/Sub。
    • PAUSED：精简版消息导出已暂停。
  • DEAD_LETTER_TOPIC_ID：要用作死信主题的现有精简版主题的 ID。该主题必须与导出订阅本身位于同一位置（可用区或区域）和项目。
  • PUBSUB_TOPIC_NAME：目标 Pub/Sub 主题的名称。示例格式：projects/my-project-id/topics/my-topic-id。

如果请求成功，则响应是 JSON 格式的精简版订阅：

{
"deliveryConfig": {
  "deliveryRequirement": "DELIVERY_REQUIREMENT",
},
"exportConfig": {
  "desiredState": "DESIRED_STATE",
  "deadLetterTopic": "projects/PROJECT_NUMBER/locations/LOCATION/topics/DEAD_LETTER_TOPIC_ID",
  "pubsubConfig": { "topic": "PUBSUB_TOPIC_NAME" }
},
"name": "projects/PROJECT_NUMBER/locations/LOCATION/subscriptions/SUBSCRIPTION_ID",
"topic": "projects/PROJECT_NUMBER/locations/LOCATION/topics/TOPIC_ID",
}

暂停或启动导出订阅

导出订阅有一项名为“所需状态”的设置，该设置有两个值之一：

  • 活跃：订阅会将精简版消息导出到 Pub/Sub。
  • 已暂停：已暂停导出精简版消息。

如需在 Google Cloud 控制台中更改所需状态，请执行以下操作：

  1. 前往精简版订阅页面。

    前往“精简版订阅”

  2. 点击精简版订阅 ID。

  3. 在精简版订阅详情页面中，点击暂停或开始。

您还可以使用 Google Cloud CLI 或 Pub/Sub Lite API 更新所需的状态。请参阅更新导出订阅。

最佳实践

本部分介绍了使用导出订阅时的一些最佳实践。

预留

我们建议将导出订阅与预留搭配使用，而不是明确设置订阅的吞吐量容量。

消息兼容性

如果 Pub/Sub Lite 消息与 Pub/Sub 不兼容，则导出订阅不会将该消息发布到 Pub/Sub。相反，它会将消息置于死信主题（如果已分配）中。如果未分配死信主题，则不兼容的消息会被直接丢弃。

将消息发布到精简版主题时，请注意以下兼容性问题：

  • 密钥。Pub/Sub 精简版键的类型为 bytes，而 Pub/Sub 排序键的类型为 string。为了保持兼容性，Pub/Sub Lite 键只能包含 UTF-8 字符。

  • 属性。消息属性具有以下要求：

    • 为了兼容，所有 Pub/Sub Lite 消息属性都必须具有一个值。Pub/Sub Lite 支持具有多个值的消息属性，但 Pub/Sub 仅支持单值属性。
    • 消息特性不得超过 Pub/Sub 消息限制，包括每条消息的最大特性数量，以及每个特性的键和值大小上限。

死信主题

如需保留和处理不兼容的消息，我们建议使用死信主题。您可以在创建导出订阅时分配死信主题，也可以更新现有的导出订阅以使用死信主题。如果订阅收到与 Pub/Sub 不兼容的消息，则会将该消息发布到死信主题。

死信主题是一个常规的 Pub/Sub Lite 主题。它必须与导出订阅位于同一位置和项目中，并且必须与源主题不同。

通常情况下，死信主题的吞吐量利用率较低。因此，我们建议为死信主题分配预留，而不是为该主题分配吞吐量。

递送错误

导出订阅会尝试将所有兼容的消息传递到目标 Pub/Sub 主题。如果消息传送失败，导出订阅会暂停。如需查找错误类别，请查看 subscription/export_status 指标。如果出现以下值，则表示存在错误：

  • PERMISSION_DENIED：权限不足，无法导出消息。
  • NOT_FOUND：未找到一项或多项资源；例如，目标主题不存在。

如需详细了解如何排查问题，请参阅对导出订阅进行问题排查。

解决错误后，导出订阅会因定期重试而自动重启。

价格

您需要为导出订阅所使用的 Pub/Sub Lite 和 Pub/Sub 资源付费。具体来说，您需要为 Pub/Sub Lite 订阅上分配的订阅吞吐量和存储空间付费，这些订阅是为 Pub/Sub Lite 主题配置的。发布到目标 Pub/Sub 主题也需要付费。请参阅 Pub/Sub 价格。

使用导出功能不会产生额外费用，并且 Pub/Sub Lite 导出订阅与标准订阅之间没有价格差异。

排查导出订阅问题

本部分介绍了有关导出订阅的一些问题排查提示。

导出订阅已暂停

如果订阅已暂停，则系统不会导出任何消息。

要检测此问题，请执行以下操作：

  • Google Cloud 控制台：查看订阅详情。如果订阅暂停，则“期望状态”和“当前状态”为 Paused。

  • 指标：subscription/export_status 指标为 PAUSED。

如需解决此问题，请启动订阅。

目标主题或死信主题已删除

如果删除附加到导出订阅的 Pub/Sub 主题，或删除死信主题，则会发生错误。

要检测此问题，请执行以下操作：

  • Google Cloud 控制台：查看订阅详情。如果主题已被删除，则当前状态为 Not found。

  • 指标：subscription/export_status 指标。如果主题已删除，则值为 NOT_FOUND。

如需解决此问题，请检查目标 Pub/Sub 主题和死信主题（如果已配置）。

如果目标 Pub/Sub 已被删除，请使用相同名称重新创建主题。导出订阅会继续发布（假设权限未更改）。

如果死信主题已被删除，请创建一个新的死信主题，并更新导出订阅以引用该主题。

不兼容的消息

如果消息与 Pub/Sub 不兼容，则不会导出。

要检测此问题，请执行以下操作：

  • 指标：subscription/unexportable_message_count 指标显示无法导出的不兼容消息的数量。

如需解决此问题，请使用死信主题来保留不兼容的消息。检查消息以确定原因，然后根据需要转换并重新发布它们。请参阅消息兼容性。

导出受到限制

要检测此问题，请执行以下操作：

  • 指标：subscription/flow_control_status 指标显示流控制原因 NO_CLIENT_TOKENS，它表示已达到每个分区的未完成字节或消息限制。在问题得到解决之前，相关导出订阅的积压输入量将会增加。

此错误有几种可能的根本原因。大多数可能的原因都发生在后端，但请检查以下内容：

  • 请确保以每个密钥小于 1 MiB/s 的速率发布共享同一密钥的精简版消息。导出订阅将精简版消息键作为 Pub/Sub 排序键写入，而 Pub/Sub 对每个排序键的限制为 1 MiB/s。超过此限制可能会导致系统受到限制。

用户无权执行此操作

Pub/Sub Lite 服务代理必须具有发布到目标 Pub/Sub 主题的权限。

要检测此问题，请执行以下操作：

  • Google Cloud 控制台：查看订阅详情。如果存在权限错误，则当前状态为 Permission denied。

  • 指标：subscription/export_status 指标为 PERMISSON_DENIED。

例如，以下情况可能会导致此错误：

  • Pub/Sub Lite 服务代理缺少将消息发布到其他项目中的目标 Pub/Sub 主题的正确权限或角色。
  • 该服务代理已从导出订阅父项目的 IAM 政策中移除。
  • Pub/Sub Lite 服务代理仍在设置中。当您在项目中创建第一个导出订阅时，系统会自动创建一个服务代理。权限错误应该会在 10 分钟内自动解决。

如需解决此问题，请检查是否已向服务代理授予正确的权限或角色。请参阅服务代理。

后续步骤

选择 Pub/Sub 或 Pub/Sub Lite。