google-pubsub

Google Cloud Pub/Sub 核心知识

完整参考：Java/Python/Go 发布者模式、Spring Cloud GCP 消费者、推送订阅处理程序，以及用于 DLT、IAM、VPC Service Controls 和监控告警的 Terraform，请参阅 advanced.md。

深入了解：使用 mcp__documentation__fetch_docs，并将 technology 设为：google-pubsub，以获取完整文档。

快速开始（模拟器）

# docker-compose.yml
services:
  pubsub:
    image: gcr.io/google.com/cloudsdktool/google-cloud-cli:emulators
    command: gcloud beta emulators pubsub start --host-port=0.0.0.0:8085
    ports:
      - "8085:8085"

docker-compose up -d

# Set emulator env
export PUBSUB_EMULATOR_HOST=localhost:8085

# Create topic and subscription
gcloud pubsub topics create orders
gcloud pubsub subscriptions create order-processor --topic=orders

# Test
gcloud pubsub topics publish orders --message='{"id":"123"}'
gcloud pubsub subscriptions pull order-processor --auto-ack

核心概念

概念	描述
Topic	用于发布消息的命名资源
Subscription	用于接收消息的命名资源
Publisher	向 topic 发送消息
Subscriber	从 subscription 接收消息
Ack Deadline	重新投递前确认消息的时限
消息保留	未确认消息的保留时长

投递类型

类型	描述	使用场景
Pull	Subscriber 请求消息	批处理、负载变化较大的场景
Push	Pub/Sub 向端点发送消息	无服务器场景、Webhook
BigQuery	直接导出到 BigQuery	分析流水线
Cloud Storage	直接导出到 GCS	数据归档

架构

Publisher ──▶ Topic ──▶ Subscription 1 (Pull) ──▶ Subscriber
                   ──▶ Subscription 2 (Push) ──▶ Cloud Run
                   ──▶ Subscription 3 ──▶ BigQuery

Node.js 发布者（@google-cloud/pubsub）

import { PubSub } from '@google-cloud/pubsub';

const pubsub = new PubSub({
  projectId: 'my-project',
});

const topic = pubsub.topic('orders');

// Publish single message
const messageId = await topic.publishMessage({
  data: Buffer.from(JSON.stringify(order)),
  attributes: {
    'correlation-id': correlationId,
    'order-type': order.type,
  },
  orderingKey: order.customerId, // For ordered delivery
});

// Batch publishing (automatic batching)
const publishOptions = {
  batching: {
    maxMessages: 100,
    maxMilliseconds: 10,
  },
};
const batchTopic = pubsub.topic('orders', publishOptions);

Node.js 消费者（Pull 订阅）

const subscription = pubsub.subscription('order-processor', {
  flowControl: {
    maxMessages: 100,
    maxExtensionMinutes: 10,
  },
  ackDeadline: 30,
});

const messageHandler = async (message) => {
  try {
    const order = JSON.parse(message.data.toString());
    const correlationId = message.attributes['correlation-id'];

    await processOrder(order);
    message.ack();
  } catch (error) {
    console.error('Processing failed:', error);
    message.nack(); // Will be redelivered
  }
};

subscription.on('message', messageHandler);
subscription.on('error', (error) => console.error('Subscription error:', error));

// Graceful shutdown
process.on('SIGTERM', async () => {
  await subscription.close();
});

死信主题

# Create DLT
gcloud pubsub topics create orders-dlq
gcloud pubsub subscriptions create orders-dlq-sub --topic=orders-dlq

# Create subscription with DLT
gcloud pubsub subscriptions create order-processor \
  --topic=orders \
  --dead-letter-topic=orders-dlq \
  --max-delivery-attempts=5

不适合使用此技能的场景

以下情况请使用其他消息传递解决方案：

• 采用 AWS 原生架构时，SQS 与 AWS 的集成更好
• 采用 Azure 原生架构时，使用 Azure Service Bus
• 需要支持重放的事件流时，使用 Dataflow 或 Kafka
• 部署在本地环境时，使用 RabbitMQ 或 ActiveMQ
• 需要多云可移植性时，使用 Kafka 或 RabbitMQ
• 需要复杂路由模式时，RabbitMQ 更灵活
• 需要符合 JMS 时，使用 ActiveMQ

反模式

反模式	不良影响	解决方案
未设置死信主题	失败的消息会丢失	为所有订阅配置 DLT
Ack Deadline 过短	导致重复处理	将时限设为长于最长处理时间
未设置重试策略	失败后立即重新投递	配置指数退避
同步发布	吞吐量低	使用批处理和异步发布
Pull 未设置流量控制	消费者不堪重负	设置 max_messages 限制
需要消息排序时未设置排序	消息处理顺序错乱	使用排序键
消息负载过大	成本更高、性能更差	使用 Cloud Storage 并通过引用访问
未遵循 IAM 最小权限原则	存在安全风险	使用仅具备最低限度角色的服务账号

快速故障排查

问题	可能原因	修复方法
未收到消息	没有订阅或 topic 错误	创建订阅并验证 topic
消息重复	Ack Deadline 已过期	延长 Ack Deadline 或加快处理速度
消息进入 DLT	超过最大投递次数	检查处理逻辑并查看 DLT
权限被拒绝	缺少 IAM 角色	授予 Publisher/Subscriber 角色
排序未生效	没有排序键或订阅配置错误	设置排序键并启用消息排序
延迟较高	批处理延迟或网络问题	缩短批处理延迟并检查网络
推送订阅失败	端点不可用或认证失败	检查端点运行状况并验证认证
积压不断增加	消费者处理速度慢	增加消费者或优化处理流程

生产环境就绪度

监控指标

指标	告警阈值
num_undelivered_messages	> 10000
oldest_unacked_message_age	> 3600s
num_outstanding_messages	> 10000
dead_letter_message_count	> 0
publish_latencies	p99 > 1s

检查清单

• IAM 角色遵循最小权限原则
• 每个组件使用独立的服务账号
• 已配置死信主题
• 已配置重试策略
• 确认应答期限设置合理
• 已设置消息保留期限
• 已配置监控告警
• 模式验证（如需要）
• 消息排序（如需要）
• VPC Service Controls（如需要）

参考文档

深入了解：使用 mcp__documentation__fetch_docs，并将 technology 设为：google-pubsub，以获取完整文档。

可用主题：basics、producers、consumers、production