将 Pub/Sub Lite 消息导出到 Pub/Sub

本文档介绍了如何设置 发送到 Pub/Sub 的 Pub/Sub Lite 消息。

以下是您可能使用此功能的一些场景：

  • 在混合使用 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 Lite 消息的图表

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

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

身份验证

导出订阅可同时访问 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 Lite 服务代理。如果您在同一项目中创建其他导出订阅，这些订阅将使用相同的服务代理。服务代理采用以下命名方案：service-<your_project_number>@gcp-sa-pubsublite.iam.gserviceaccount.com。

创建服务代理时，系统会为其授予向与导出订阅位于同一项目中的所有 Pub/Sub 和 Pub/Sub Lite 主题发布消息的权限。如果目标 Pub/Sub 主题与导出订阅位于不同的项目中，您必须通过添加 Pub/Sub 发布商角色 (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：目标 Pub/Sub 的名称 主题以添加 IAM 政策绑定。
  • 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 主题以接收导出的 Lite 消息。

  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：订阅会将 Lite 消息导出到 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 精简版支持的位置的名称。
  • 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：订阅会将 Lite 消息导出到 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：订阅会将 Lite 消息导出到 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：订阅会将 Lite 消息导出到 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",
}

暂停或启动导出订阅

导出订阅有一个名为所需状态的设置，该设置具有以下两个值之一：

  • 有效：订阅会将 Lite 消息导出到 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 精简版密钥只能包含 UTF-8 字符。

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

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

死信主题

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

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

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

递送错误

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

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

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

解决错误后，由于系统会定期重试，导出订阅会自动重启。

价格

您需要为导出订阅使用的 Pub/Sub Lite 和 Pub/Sub 资源付费。具体来说，我们会向您收取费用 为集群上分配的订阅吞吐量和存储空间 Pub/Sub Lite 订阅，该订阅已配置为 Pub/Sub 精简版主题。如果将内容发布到 目标 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/秒。导出订阅将精简版消息键写入 Pub/Sub 排序键，Pub/Sub 的排序为 1 MiB/s 限制。超出此限制可能会导致节流。

用户无权执行此操作

Pub/Sub Lite 服务代理必须拥有向目标 Pub/Sub 主题发布消息的权限。

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

  • Google Cloud 控制台：查看 订阅详情。如果有 权限错误，则 Current state 为 Permission denied。

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

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

  • Pub/Sub Lite 服务代理缺少向其他项目中的目标 Pub/Sub 主题发布消息的正确权限或角色。
  • 从导出的 IAM 政策中移除了服务代理 订阅父级项目。
  • Pub/Sub Lite 服务代理仍在设置中。服务 代理。 资源。权限错误应该会在 10 天内自动解决 分钟。

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

后续步骤

选择 Pub/Sub 或 Pub/Sub Lite。