3. Pub/Sub实战

前两节我们理解了 Pub/Sub 的“主题—订阅—消息”模型，本节把它落到可执行的操作上。Pub/Sub 的核心对象只有三个：主题（Topic）是消息的入口，发布者只认主题；订阅（Subscription）是消息的出口，每个订阅独立保存一份待投递的消息副本；消息（Message）由 data 与 attributes 组成，data 是字节串，attributes 是键值对元数据。理解这一点，后面的命令和代码就都有了归属。

一、用 gcloud 创建主题与订阅

先确认项目与身份，再创建资源。以下命令中的 my-project、orders-topic 均为示例值，实际使用时替换为自己的项目与命名。

gcloud config set project my-project

# 创建主题
gcloud pubsub topics create orders-topic

# 创建订阅：拉取模式，消息保留 7 天，确认截止 60 秒
gcloud pubsub subscriptions create orders-sub \
  --topic=orders-topic \
  --ack-deadline=60 \
  --message-retention-duration=7d

# 查看资源
gcloud pubsub topics list
gcloud pubsub subscriptions list
gcloud pubsub subscriptions describe orders-sub

这里有两个参数值得展开。--ack-deadline 是订阅者收到消息后必须在此时间内 ack（确认）的期限，超时未确认，消息会被重新投递；--message-retention-duration 决定未被确认的消息最多保留多久，超过即被丢弃。二者共同决定了“消息能重试多久”。

如果希望消息被多个独立消费者各消费一份，就为同一主题建多个订阅；如果希望多个消费者分摊同一批消息，则在同一订阅下挂多个订阅者客户端即可，Pub/Sub 会自动做负载均衡。这是初学者最容易混淆的一点。

二、发布消息：gcloud 与客户端库

命令行适合快速验证。发布一条带属性的消息：

gcloud pubsub topics publish orders-topic \
  --message='{"orderId":"A1001","amount":199}' \
  --attribute=region=cn-north,type=create

生产环境更常用客户端库。以 Python 为例，发布端的关键是批量发送与异步 future：

from google.cloud import pubsub_v1

publisher = pubsub_v1.PublisherClient()
topic_path = publisher.topic_path("my-project", "orders-topic")

future = publisher.publish(
    topic_path,
    data=b'{"orderId":"A1001","amount":199}',
    region="cn-north",
    type="create",
)
print("message id:", future.result())

publish() 返回的是 future，客户端会按批量和延迟自动聚合消息，调用 result() 才真正等待服务端返回 message id。发布端要关注的是吞吐与背压：批量越大吞吐越高，但单条延迟也越大，需要按业务在 BatchSettings 中权衡。

三、订阅消息：拉取与流式拉取

拉取模式最直观，适合批处理和调试：

subscriber = pubsub_v1.SubscriberClient()
sub_path = subscriber.subscription_path("my-project", "orders-sub")

resp = subscriber.pull(subscription=sub_path, max_messages=10)
for m in resp.received_messages:
    print(m.message.data, m.message.attributes)
    subscriber.acknowledge(subscription=sub_path, ack_ids=[m.ack_id])

注意 ack_id 与消息一一对应，必须显式 ack，否则消息会在 ack deadline 后重投。生产上更推荐流式拉取（StreamingPull），它保持长连接、自动续期 ack deadline，并用回调处理消息：

def callback(message):
    print(message.data)
    message.ack()

future = subscriber.subscribe(sub_path, callback=callback)
future.result()

回调中若抛异常而不 ack，消息会被重投；若业务需要延迟重试，可调用 message.nack() 主动让服务端尽快重投，而不是干等 ack deadline 超时。

四、监控积压与延迟

积压和延迟是 Pub/Sub 运维的两个核心信号，它们不是同一个东西：积压看“有多少条没被确认”，延迟看“最老的那条等了多久”。一个订阅可能积压条数不多，但最老消息年龄很大，说明消费端卡住了；也可能积压条数很多，但年龄很小，说明只是瞬时流量高峰。

在 Cloud Monitoring 中重点看三个指标：subscription/num_undelivered_messages（未投递/未确认消息数，即积压）、subscription/oldest_unacked_message_age（最老未确认消息年龄，即延迟）、以及 topic/sent_message_count 与 subscription/acked_message_count 的对比。若 ack 速率长期低于发送速率，积压必然持续增长。

命令行也能快速查看积压：

gcloud pubsub subscriptions describe orders-sub \
  --format="value(name,messageRetentionDuration,ackDeadlineSeconds)"

更细的积压数值建议在 Monitoring 控制台或通过 Monitoring API 查询，因为 describe 返回的是配置而非实时指标。

五、常见问题与判断依据

• 消息重复：Pub/Sub 至少一次投递，重复是正常的。消费端必须幂等，例如用 orderId 做去重键。
• 消息一直重投：检查 ack deadline 是否太短、回调是否抛异常、是否忘记 ack。流式拉取会自动续期，但处理时间过长仍可能超时。
• 积压持续增长：先看 ack 速率与发送速率，再确认消费者实例数与单条处理耗时。必要时增加订阅者并发或优化下游。
• 延迟高但积压低：通常是少量“毒消息”反复重试，检查是否有消息处理失败后未 nack 也未 ack。