pubsub-basics

GCP Pub/Sub 基础知识

本技能提供通过 gcloud CLI 操作 Google Cloud Pub/Sub 的安全、可重复指南。内容涵盖核心消息传递概念、操作资源所需的 IAM 角色、创建主题和订阅的分步教程，以及用于防止破坏性或含糊操作的运维防护措施。在为事件驱动系统、流式处理管道或异步服务集成配置、检查或操作 Pub/Sub 基础设施时，请遵循本指南。

概述

Google Cloud Pub/Sub 是一项全托管的异步消息服务，可将消息生产者与消息使用者解耦。生产者发布消息时无需知道由谁使用这些消息，而使用者接收消息时也无需知道消息由谁产生。这种解耦能够支持具备弹性、可水平扩展的事件驱动架构。

核心概念

• 主题 — 生产者（发布者）向其发送消息的命名资源。主题是数据流入系统的入口点。主题还可以选择对传入消息强制执行 架构。

• 订阅 — 表示来自单个主题的消息流并将其传递给订阅应用的命名资源。每个订阅都会收到该订阅创建后发布到主题的每条消息各自独立的副本。一个主题可以有多个订阅；这就是扇出机制。

• 拉取传递 — 订阅应用按自己的计划主动从订阅请求消息（使用客户端库、CLI 或流式拉取连接）。拉取模式非常适合高吞吐量的批处理使用者，以及能够自行控制并发的工作进程。成功处理消息后，必须显式确认（ack）消息。

• 推送传递 — Pub/Sub 主动将每条消息作为 HTTPS POST 请求发送到配置的 端点（例如 Cloud Run 服务、Cloud Functions 函数，或任何可通过公网访问的 HTTPS URL）。端点通过返回成功的 HTTP 状态码（102、200、201、202 或 204）来确认消息。推送模式非常适合无服务器使用者和 webhook 风格的集成。

• 确认（ack）和确认期限 — 使用者必须确认消息，以表示已成功处理。如果消息未在订阅的 确认期限内得到确认，Pub/Sub 会重新传递该消息。这种至少传递一次的模式意味着使用者必须设计为 幂等。

• 死信主题（DLT） — 当订阅传递消息失败次数超过配置的最大次数（--max-delivery-attempts）后，消息会被转发到的主题。死信机制会隔离“毒性”消息，使其不再阻塞正常流量，并可供之后检查或重新处理。Pub/Sub 服务账号必须有权向死信主题发布消息，并订阅源订阅。

• 架构 — 可选的版本化定义（Avro 或 Protocol Buffer），用于约束发布到主题的消息结构。如果主题关联了架构和编码方式（JSON 或 BINARY），Pub/Sub 就会验证每条已发布的消息，并拒绝不符合要求的负载。架构可为生产者和使用者提供契约，防止格式错误的数据进入管道。

前置条件

执行任何操作前，请确认以下条件均已满足。

1. 启用 Pub/Sub API。 目标项目必须已启用 pubsub.googleapis.com 服务。

   gcloud services enable pubsub.googleapis.com --project=PROJECT_ID
   

2. 安装并更新 gcloud CLI。 本技能中的操作假定使用近期版本的 Google Cloud CLI。请确认已安装且为最新版本。

   gcloud version
   gcloud components update --quiet
   

3. 授权 CLI。 使用具备下文所述必要角色的身份（用户或服务账号）进行身份验证。

   gcloud auth login
   gcloud config set project PROJECT_ID
   

   对于自动化或非交互式环境，优先使用由专用服务账号支持的 Application Default Credentials：

   gcloud auth activate-service-account --key-file=KEY_FILE_PATH
   

4. 确认当前项目和账号。 修改资源前，务必确认执行上下文。

   gcloud config list --format="value(core.project,core.account)"
   

关键 IAM 角色

授予任务所需的最小权限。通过本技能执行操作的主体只能持有与其职责相符的角色。

角色	角色 ID	用途
Pub/Sub 管理员	roles/pubsub.admin	完全控制：创建、更新和删除主题、订阅和架构，并管理 IAM 政策。仅限授予负责资源配置的身份。
Pub/Sub 发布者	roles/pubsub.publisher	向主题发布消息。授予生产者服务。
Pub/Sub 订阅者	roles/pubsub.subscriber	从订阅中消费并确认消息。授予消费者服务。
Pub/Sub 查看者	roles/pubsub.viewer	只读访问权限，用于查看主题、订阅、架构和配置。授予负责检查和审计的身份。
服务账号用户	roles/iam.serviceAccountUser	允许某个身份以服务账号身份行事（模拟该服务账号）。配置经过身份验证的推送订阅并使用服务账号身份令牌调用服务时，需要此角色。

补充说明：

• 调用私有服务（例如经过身份验证的 Cloud Run 端点）的推送订阅，要求推送服务账号在目标服务上拥有调用者角色（例如 roles/run.invoker）；此外，配置者身份还必须在该服务账号上拥有 roles/iam.serviceAccountUser。
• 死信投递要求 Pub/Sub 服务代理（[email protected]）在死信主题上拥有 roles/pubsub.publisher，并在源订阅上拥有 roles/pubsub.subscriber。

复合资源部署步骤

如果需要同时配置新的架构绑定主题和订阅，请使用此复合步骤。它会在单个执行块中协调依赖资源（架构 -> 主题 -> 订阅）的配置、验证和创建，从而尽量减少 Shell 交互轮次，并避免产生不完整的配置。

# Run as a single copy-pasteable execution block
(
  set -e

  # 1. Configuration Parameters
  PROJECT_ID="PROJECT_ID"
  SCHEMA_ID="SCHEMA_ID"
  SCHEMA_TYPE="avro"                 # avro or protocol-buffer
  SCHEMA_FILE="PATH_TO_SCHEMA_FILE"   # Local path to schema definition file
  ENCODING="json"                    # json or binary
  TOPIC_ID="TOPIC_ID"
  SUBSCRIPTION_ID="SUBSCRIPTION_ID"
  ACK_DEADLINE=60                    # Acknowledgement deadline in seconds

  echo "=== [1/5] Validating Schema Definition ==="
  gcloud pubsub schemas validate-schema \
    --type="${SCHEMA_TYPE}" \
    --definition-file="${SCHEMA_FILE}" \
    --project="${PROJECT_ID}"

  echo "=== [2/5] Creating Schema ==="
  gcloud pubsub schemas create "${SCHEMA_ID}" \
    --type="${SCHEMA_TYPE}" \
    --definition-file="${SCHEMA_FILE}" \
    --project="${PROJECT_ID}" \
    --quiet

  echo "=== [3/5] Creating Topic ==="
  gcloud pubsub topics create "${TOPIC_ID}" \
    --schema="${SCHEMA_ID}" \
    --message-encoding="${ENCODING}" \
    --project="${PROJECT_ID}" \
    --quiet

  echo "=== [4/5] Creating Subscription ==="
  gcloud pubsub subscriptions create "${SUBSCRIPTION_ID}" \
    --topic="${TOPIC_ID}" \
    --ack-deadline="${ACK_DEADLINE}" \
    --project="${PROJECT_ID}" \
    --quiet

  echo "=== [5/5] Verifying Resources ==="
  gcloud pubsub schemas describe "${SCHEMA_ID}" --project="${PROJECT_ID}"
  gcloud pubsub topics describe "${TOPIC_ID}" --project="${PROJECT_ID}"
  gcloud pubsub subscriptions describe "${SUBSCRIPTION_ID}" --project="${PROJECT_ID}"
)

架构约束：“JSON”是一种消息编码，而不是架构类型

这一点几乎人人都会弄错。Google Cloud Pub/Sub 只有两种架构类型： AVRO 和 PROTOCOL_BUFFER。没有 JSON 架构类型，执行 gcloud pubsub schemas create --type=json 会失败。

当任务要求提供“JSON 架构”或“严格 JSON 验证”时，意思是主题根据所附架构，以 JSON 编码验证消息。JSON 是你为主题设置的 --message-encoding，而非架构的类型。架构定义通常使用 AVRO 编写。

因此，“严格 JSON 架构”的要求可分为两部分：

1. 附加到主题上的架构（数据契约）。
2. 将主题的 schemaSettings.encoding 设置为 JSON（以 JSON 格式验证消息）。

正确做法

SCHEMA="orders-schema"
TOPIC="orders-topic"

# 1. Create the schema. Type is AVRO; the definition is written in Avro JSON.
gcloud pubsub schemas create "$SCHEMA" \
  --type=AVRO \
  --definition='{"type":"record","name":"Order","fields":[
    {"name":"id","type":"string"},
    {"name":"amount","type":"double"}
  ]}'

# 2. Create the topic and bind the schema with JSON MESSAGE ENCODING.
#    --message-encoding=json is the part that makes it a "JSON schema".
gcloud pubsub topics create "$TOPIC" \
  --schema="$SCHEMA" \
  --message-encoding=json

# 3. Verify the contract: encoding must be JSON.
gcloud pubsub topics describe "$TOPIC" --format="value(schemaSettings.encoding)"   # -> JSON

如果主题已存在，请使用 gcloud pubsub topics update "$TOPIC" --schema="$SCHEMA" --message-encoding=json 绑定架构和编码。

除非任务明确要求二进制，否则默认使用 JSON 编码。 只有在需求要求时，才使用 --message-encoding=binary。选择二进制（或省略编码）是“JSON 架构”要求最常被判定为未满足的原因，因为架构看似已设置，但消息并未经过 JSON 验证。

故障排查

• “架构是 AVRO，但我需要 JSON”： 这是概念分类错误。AVRO 是架构类型；JSON 是消息编码。保留 AVRO 架构，并设置主题的 --message-encoding=json。不要尝试更改架构类型。
• 编码显示为空 / BINARY： 创建主题时未指定 --message-encoding=json（或未指定 --schema）。请使用 gcloud pubsub topics update --schema=... --message-encoding=json 重新绑定。

死信主题：策略配置在消费端订阅上

死信策略应配置在从主主题拉取消息的订阅上，而不是死信主题上，也不是死信主题的订阅上。这里几乎所有失败都源于两个错误：将策略绑定到错误的订阅，以及忘记指定 --max-delivery-attempts。

若要在投递失败 N 次后转入死信：

DLQ_TOPIC="orders-dlq"
SUB="orders-sub"          # the subscription that consumes the MAIN topic
PROJECT="$(gcloud config get-value project)"

# 1. Create the dead-letter topic.
gcloud pubsub topics create "$DLQ_TOPIC" --project="$PROJECT"

# 2. Attach the dead-letter policy to the CONSUMING subscription.
#    --max-delivery-attempts is required; without it there is no dead-lettering.
gcloud pubsub subscriptions update "$SUB" \
  --project="$PROJECT" \
  --dead-letter-topic="$DLQ_TOPIC" \
  --max-delivery-attempts=5

# 3. Verify the policy is on the right subscription.
gcloud pubsub subscriptions describe "$SUB" \
  --project="$PROJECT" \
  --format="value(deadLetterPolicy.deadLetterTopic, deadLetterPolicy.maxDeliveryAttempts)"

请向 Pub/Sub 服务代理授予 IAM 权限，否则死信转发会静默失败。 服务代理 service-<project-number>@gcp-sa-pubsub.iam.gserviceaccount.com 需要在死信主题上拥有 roles/pubsub.publisher 角色，并在订阅上拥有 roles/pubsub.subscriber 角色。否则，即使策略看起来配置正确，消息也不会转发到 DLQ。

常见错误

• 策略配置在错误的订阅上。 策略必须配置在消费源主题的订阅上。将死信策略绑定到死信主题的订阅上没有任何作用。
• 省略 --max-delivery-attempts。 有效范围为 5 到 100。如果任务要求“恰好尝试 N 次”，请明确传入 N。

创建主题和订阅

以下所有命令都使用大写占位符。执行前，请将 TOPIC_ID、 SUBSCRIPTION_ID、PROJECT_ID、ENDPOINT 等占位符替换为具体值。每条命令都会通过 --project=PROJECT_ID 明确指定项目，并使用 --quiet 以非交互方式运行。

1. 创建主题

gcloud pubsub topics create TOPIC_ID \
  --project=PROJECT_ID \
  --quiet

若要创建一个强制设置消息保留时长的主题（使订阅也能在该时间窗口内重放消息）：

gcloud pubsub topics create TOPIC_ID \
  --message-retention-duration=7d \
  --project=PROJECT_ID \
  --quiet

若要将主题关联到现有架构：

gcloud pubsub topics create TOPIC_ID \
  --schema=SCHEMA_ID \
  --message-encoding=json \
  --project=PROJECT_ID \
  --quiet

2. 创建 Pull 订阅

Pull 订阅允许消费者按需获取消息。--ack-deadline 控制消费者必须在多长时间内确认消息，超时后消息将重新投递。

gcloud pubsub subscriptions create SUBSCRIPTION_ID \
  --topic=TOPIC_ID \
  --ack-deadline=60 \
  --project=PROJECT_ID \
  --quiet

若要关联死信主题并限制重新投递次数：

gcloud pubsub subscriptions create SUBSCRIPTION_ID \
  --topic=TOPIC_ID \
  --ack-deadline=60 \
  --dead-letter-topic=DEAD_LETTER_TOPIC_ID \
  --max-delivery-attempts=5 \
  --project=PROJECT_ID \
  --quiet

若要使用基于消息属性的服务端过滤器，限制此订阅接收的消息（请参阅下文“数据缩减”）：

gcloud pubsub subscriptions create SUBSCRIPTION_ID \
  --topic=TOPIC_ID \
  --message-filter='attributes.eventType = "ORDER_CREATED"' \
  --project=PROJECT_ID \
  --quiet

3. 创建 Push 订阅

Push 订阅会将每条消息发送到 HTTPS ENDPOINT。端点必须返回成功状态，才能确认消息。

gcloud pubsub subscriptions create SUBSCRIPTION_ID \
  --topic=TOPIC_ID \
  --push-endpoint=ENDPOINT \
  --ack-deadline=60 \
  --project=PROJECT_ID \
  --quiet

若要创建指向私有服务的经过身份验证的 Push 订阅，请关联 Push 身份验证服务账号，以便 Pub/Sub 发送已签名的 OIDC 令牌：

gcloud pubsub subscriptions create SUBSCRIPTION_ID \
  --topic=TOPIC_ID \
  --push-endpoint=ENDPOINT \
  --push-auth-service-account=PUSH_SERVICE_ACCOUNT_EMAIL \
  --ack-deadline=60 \
  --project=PROJECT_ID \
  --quiet

4. 验证创建的资源

创建后，请先确认配置，再依赖这些资源。

gcloud pubsub topics describe TOPIC_ID \
  --project=PROJECT_ID

gcloud pubsub subscriptions describe SUBSCRIPTION_ID \
  --project=PROJECT_ID

资源命名与验证清单

完成任何 Pub/Sub 资源配置任务前，请按此清单逐项检查，确保所有资源都符合预期：

• 验证资源名称是否符合要求：对照提示词或规范，核实 Schema、Topic 和 Subscription 的确切名称。不要臆测名称。
• 验证是否已应用标签：如果提示词要求添加标签（例如 windtunnel-eval），请确认 Topic 和 Subscription 上都已添加。
• 验证资源是否存在：对 Schema、Topic 和 Subscription 运行 describe 命令，并确认输出显示 state: ACTIVE（对于 Schema），且配置正确。

核心原则

以下原则为强制要求。每次执行 Pub/Sub 操作时都必须遵守。

明确验证命令（强制要求）

• 运行任何会修改状态的命令前，先用通俗语言重述该命令将执行的操作以及目标资源。确认资源名称、项目和操作效果。
• 如果子命令支持 --dry-run，请使用；或者先运行只读的 describe／list 等效命令，确认当前状态。
• 绝不执行效果无法精确预测的命令。如果结果不明确，请先停止并验证。

减少数据量

尽量减少获取、处理和返回的数据量。读取大量未经筛选的数据会拖慢操作、增加成本并产生冗余信息。

• --limit — 列出或拉取时限制结果数量。例如，gcloud pubsub subscriptions pull SUBSCRIPTION_ID --limit=10。
• 服务端筛选 — 创建 Subscription 时应用 --message-filter，避免传递不需要的消息，而不是事后再筛选。在支持的情况下，对 list 命令使用 --filter，在服务端缩小结果范围。
• 字段投影 — 使用 --format="value(...)" 或 --format="table(...)" 仅请求所需字段，而不是输出完整资源信息。

项目和位置范围限定

• 始终在每条命令中附加 --project=PROJECT_ID。不要只依赖当前的 gcloud config 项目；明确指定范围可避免误操作其他项目。
• 对支持位置或区域端点的资源，请明确指定位置，不要依赖默认值。

执行限制

• 非交互模式 — 附加 --quiet（或 -q）以抑制交互式提示，确保命令行为确定且适合自动化。不得用此选项绕过下方“安全”部分所述的人工审批要求。
• 单条命令 — 每次只执行一条独立的 gcloud 命令。不要在一次调用中串联无关操作。
• 禁止使用原始 Shell 运算符 — 不要使用 Shell 管道（|）、重定向（>、>>）、命令替换（$(...)）、后台运算符（&）或 &&／; 链式操作来组合 Pub/Sub 命令。每条命令都必须独立执行并可单独审计。

安全与防护措施

注意

Pub/Sub 操作可能具有破坏性且不可逆。删除 Topic 或 Subscription 会永久丢弃尚未投递的消息，并中断所有绑定到它的生产者和消费者。更改 IAM policy 可能会在不知情的情况下切断生产服务的访问权限。除非已明确证明不会影响生产环境，否则应将每项修改操作都视为会影响生产环境。

禁止操作（拒绝列表）

未经明确且有记录的人类批准，不得执行以下操作。若任务似乎需要执行其中某项操作，请先停止并请求批准；说明确切命令及其影响范围。

• 删除主题 — gcloud pubsub topics delete ...。此操作会销毁主题并解除所有订阅与该主题的关联。
• 删除订阅 — gcloud pubsub subscriptions delete ...。 此操作会永久丢弃该订阅中所有尚未确认的消息。
• 删除或修改架构 — gcloud pubsub schemas delete ...，或可能导致正在处理的生产者失效的架构修订变更。
• 以破坏性方式清除／转移消息 — 执行 gcloud pubsub subscriptions seek ...，转移到会丢弃当前积压消息的时间戳或快照。
• 更改关键 IAM 绑定 — 在生产主题或订阅上添加、移除或替换 roles/pubsub.admin、roles/pubsub.publisher 或 roles/pubsub.subscriber 绑定，或任何会替换整个策略的 set-iam-policy 操作。
• 解除主题关联 — gcloud pubsub topics detach-subscription ...，此操作会立即停止向受影响的订阅投递消息。

始终安全的操作

只读检查命令（describe、list，以及用于诊断且有数量限制、不自动确认的 pull）是安全的，建议用于验证，但须遵守上述数据缩减原则。

故障排查／遇到以下情况时的处理方法……

……命令因权限错误而失败（PERMISSION_DENIED / 403）

1. 确认当前身份：gcloud config list --format="value(core.account)"。
2. 确认资源上的角色与操作相符（见 IAM 表）。例如，发布消息需要 roles/pubsub.publisher；消费消息需要 roles/pubsub.subscriber。
3. 若要推送到私有端点，请确认推送服务账号在目标服务上拥有调用方角色，并且配置身份在该服务账号上拥有 roles/iam.serviceAccountUser。
4. 确认已启用 API：gcloud services list --enabled --filter="pubsub" --project=PROJECT_ID。

……订阅的积压消息不断增加（消息消费速度不够快）

1. 在 Cloud Monitoring 指标中检查积压量和最早未确认消息的时长（subscription/num_undelivered_messages、subscription/oldest_unacked_message_age）。
2. 对于拉取式消费者，增加消费者实例数量，或提高每个实例的并发数，使总吞吐量超过发布速率。
3. 确认消费者成功确认消息；处理失败会导致重新投递并使积压量增大。
4. 考虑是否应使用 --message-filter 减少投递量，或该主题是否被过度扇出。

……消息持续处于未确认状态（反复重新投递）

1. 确认消费者在成功处理后调用了 acknowledge。
2. 检查 --ack-deadline。如果处理通常耗时超过此时限，请延长时限或以编程方式延长确认期限；否则，Pub/Sub 会在消费者完成处理之前重新投递消息。
3. 确保消费者具备幂等性——至少一次投递保证意味着可能出现重复消息，而非幂等处理程序会使问题加剧。
4. 如果某些特定的“毒消息”始终无法处理成功，请通过 --max-delivery-attempts 配置死信主题，将其隔离，而不是无限期重新投递。

...架构验证失败（已发布的消息被拒绝）

1. 确认主题绑定的架构和编码： gcloud pubsub topics describe TOPIC_ID --project=PROJECT_ID。
2. 发布前，使用架构验证示例消息： gcloud pubsub schemas validate-message ...。
3. 确认生产者使用主题要求的编码序列化数据 （JSON 与 BINARY）；即使数据在结构上有效，编码不匹配也会导致拒绝。
4. 如果确实需要更改架构，请提交新的兼容架构版本，而不是临时修改生产者——并将删除架构视为需要审批的禁止操作。

参考目录

如需更深入、针对具体任务的详细信息，请参阅以下参考文档：

• references/cli-usage.md — 全面的 gcloud pubsub 命令参考，涵盖发布、拉取、确认、定位、快照和输出格式示例。
• references/iam-security.md — 详细的 IAM 角色 和策略指南、最小权限模式、推送传递的服务账号配置，以及死信权限设置。
• references/client-libraries.md — Google Cloud 官方客户端库的相关指引，介绍如何以编程方式发布和消费消息，并说明何时应优先使用 SDK 代码而不是 CLI。