Google Cloud Pub/Sub 消息队列与事件驱动架构实战指南

为什么选择 Google Cloud Pub/Sub

在微服务架构和分布式系统盛行的今天，消息队列已经成为后端架构的标配组件。Google Cloud Pub/Sub 作为 Google Cloud Platform 提供的全托管消息中间件，以其高可用性、自动扩展和全球覆盖能力，在众多消息队列方案中独树一帜。

与传统的消息队列（如 RabbitMQ、Apache Kafka）不同，Pub/Sub 完全免运维——你不需要管理 Broker 集群、不需要操心分区再平衡、也不需要为磁盘空间发愁。Google 承诺 99.95% 的可用性 SLA，并且消息持久化最长可达 7 天。更重要的是，Pub/Sub 原生支持全球范围的消息传递，不局限于单一区域，这在构建跨国应用时极具吸引力。

本文将从一个实际项目出发，带你从零搭建一套完整的 Pub/Sub 消息系统，涵盖主题创建、消息发布与订阅、死信队列、与 Cloud Run 的集成，以及生产环境的最佳实践。

Google Cloud Pub/Sub 架构示意图

Pub/Sub 核心概念速览

在动手之前，先理解 Pub/Sub 的几个核心概念：

概念	说明	类比（Kafka）
Topic（主题）	消息的逻辑分类管道	Topic
Subscription（订阅）	从 Topic 拉取消息的通道	Consumer Group
Message（消息）	包含数据和属性的通信单元	Record
Publisher（发布者）	向 Topic 发送消息的客户端	Producer
Subscriber（订阅者）	从 Subscription 接收消息的客户端	Consumer
Ack（确认）	订阅者告知系统消息已处理完毕	Commit Offset
Dead Letter Topic	超过最大重试次数的消息转发目标	DLQ

Pub/Sub 的投递模型是”至少一次”（at-least-once），这意味着同一条消息可能被多次投递，所以你的消费者代码需要具备幂等性。同时 Pub/Sub 不保证消息顺序，除非你使用分区主题（ordering key）。

准备工作：项目初始化与认证

启用 Pub/Sub API 并配置 SDK

首先确保你已经在 GCP 项目中启用了 Pub/Sub API。然后安装 Google Cloud SDK 和 Pub/Sub 客户端库：


1
2
3
4
5
6
7
8
9
10
# 安装 gcloud CLI（如果尚未安装）

# curl https://sdk.cloud.google.com | bash

# exec -l $SHELL

# gcloud init



# 启用必要 API

gcloud services enable pubsub.googleapis.com



# 安装 Python 客户端库

pip install google-cloud-pubsub

创建服务账号与凭据

生产环境中不要使用用户账号，应该创建专用的服务账号：


1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
# 创建服务账号

gcloud iam service-accounts create pubsub-sa \

  --display-name="Pub/Sub Service Account"



# 授予发布权限

gcloud projects add-iam-policy-binding YOUR_PROJECT_ID \

  --member="serviceAccount:pubsub-sa@YOUR_PROJECT_ID.iam.gserviceaccount.com" \

  --role="roles/pubsub.publisher"



# 授予订阅权限

gcloud projects add-iam-policy-binding YOUR_PROJECT_ID \

  --member="serviceAccount:pubsub-sa@YOUR_PROJECT_ID.iam.gserviceaccount.com" \

  --role="roles/pubsub.subscriber"



# 下载 JSON 密钥

gcloud iam service-accounts keys create ~/pubsub-sa-key.json \

  --iam-account=pubsub-sa@YOUR_PROJECT_ID.iam.gserviceaccount.com



# 设置环境变量

export GOOGLE_APPLICATION_CREDENTIALS=~/pubsub-sa-key.json

实战一：创建主题和订阅

我们以一个电商订单处理系统为例，构建一个完整的消息流。首先创建一个订单主题和对应的订阅。

通过 gcloud 命令行创建


1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
# 设置项目 ID

PROJECT_ID="your-project-id"

TOPIC_ID="order-events"

SUBSCRIPTION_ID="order-events-sub"



# 创建主题

gcloud pubsub topics create $TOPIC_ID \

  --project=$PROJECT_ID \

  --message-retention-duration=3d \

  --labels=env=prod,service=order



# 创建订阅（拉取模式）

gcloud pubsub subscriptions create $SUBSCRIPTION_ID \

  --topic=$TOPIC_ID \

  --project=$PROJECT_ID \

  --ack-deadline=60 \

  --message-retention-duration=7d \

  --expiration-period=never \

  --enable-exactly-once-delivery=false

参数说明：

1
--ack-deadline

：消费者必须在 60 秒内确认消息，超时则消息可被重新投递
1
--message-retention-duration

：未确认的消息保留时间
1
--expiration-period=never

：订阅不会因无活跃消费者而自动删除
1
--enable-exactly-once-delivery

：精确一次交付（需要更高成本，按需开启）

通过 Python SDK 创建


1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
from google.cloud import pubsub_v1



project_id = "your-project-id"

topic_id = "order-events"



# 创建 Publisher 客户端

publisher = pubsub_v1.PublisherClient()

topic_path = publisher.topic_path(project_id, topic_id)



# 创建主题（如果不存在）

try:

    topic = publisher.create_topic(

        request={

            "name": topic_path,

            "labels": {"env": "prod", "service": "order"},

            "message_retention_duration": {"seconds": 259200},  # 3天

        }

    )

    print(f"主题已创建: {topic.name}")

except Exception as e:

    print(f"主题已存在或创建失败: {e}")



# 创建订阅（拉取模式）

subscriber = pubsub_v1.SubscriberClient()

subscription_id = "order-events-sub"

subscription_path = subscriber.subscription_path(project_id, subscription_id)



try:

    subscription = subscriber.create_subscription(

        request={

            "name": subscription_path,

            "topic": topic_path,

            "ack_deadline_seconds": 60,

            "message_retention_duration": {"seconds": 604800},  # 7天

            "enable_exactly_once_delivery": False,

        }

    )

    print(f"订阅已创建: {subscription.name}")

except Exception as e:

    print(f"订阅已存在或创建失败: {e}")

Python Pub/Sub SDK 开发

实战二：消息发布（Publisher）

发布者负责向 Topic 发送消息。每条消息包含一个字节数据载荷和可选的属性：


1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
import json

from datetime import datetime

from google.cloud import pubsub_v1



project_id = "your-project-id"

topic_id = "order-events"



publisher = pubsub_v1.PublisherClient()

topic_path = publisher.topic_path(project_id, topic_id)



def publish_order_event(event_type: str, order_data: dict) -> str:

    """

    发布订单事件到 Pub/Sub

    """

    message = {

        "event_type": event_type,

        "order_id": order_data["order_id"],

        "user_id": order_data["user_id"],

        "amount": order_data["amount"],

        "timestamp": datetime.utcnow().isoformat() + "Z",

        "data": order_data,

    }



    data_bytes = json.dumps(message, ensure_ascii=False).encode("utf-8")



    # 使用 ordering key 保证同一订单的消息顺序

    ordering_key = order_data["order_id"]



    # 添加自定义属性用于过滤

    future = publisher.publish(

        topic_path,

        data=data_bytes,

        event_type=event_type,

        source_service="order-service",

        environment="production",

        ordering_key=ordering_key,

    )



    message_id = future.result()

    print(f"消息已发布: {message_id}, 事件类型: {event_type}, 订单: {order_data['order_id']}")

    return message_id





# 使用示例

if __name__ == "__main__":

    # 订单创建事件

    order_created = {

        "order_id": "ORD-20260711-001",

        "user_id": "USER-88888",

        "amount": 299.00,

        "items": [

            {"sku": "TECH-001", "name": "无线蓝牙耳机", "quantity": 1, "price": 299.00}

        ],

        "shipping_address": "北京市朝阳区建国路88号",

    }

    publish_order_event("order.created", order_created)



    # 订单支付成功事件

    order_paid = {

        "order_id": "ORD-20260711-001",

        "user_id": "USER-88888",

        "amount": 299.00,

        "payment_method": "wechat",

        "payment_id": "PAY-20260711-ABC123",

    }

    publish_order_event("order.paid", order_paid)

消息属性（attributes）的妙用：

你可以设置
1
event_type

属性，让订阅端据此分流处理
使用
1
ordering_key

保证具有相同 Key 的消息被顺序投递
属性可用于过滤订阅（Filter Subscription），减少不必要的数据传输

实战三：消息订阅（Subscriber）

订阅者从 Subscription 中拉取消息并处理。下面是订单处理服务的完整实现：


1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
import json

import time

from concurrent.futures import TimeoutError

from google.cloud import pubsub_v1



project_id = "your-project-id"

subscription_id = "order-events-sub"



subscriber = pubsub_v1.SubscriberClient()

subscription_path = subscriber.subscription_path(project_id, subscription_id)



# 消息处理回调

def callback(message: pubsub_v1.subscriber.message.Message) -> None:

    """

    处理接收到的消息

    """

    try:

        # 解析消息数据

        data = json.loads(message.data.decode("utf-8"))

        event_type = message.attributes.get("event_type", "unknown")

        order_id = data.get("order_id", "unknown")



        print(f"[{time.strftime('%H:%M:%S')}] 收到事件: {event_type}, 订单: {order_id}")



        # 根据事件类型分流处理

        if event_type == "order.created":

            handle_order_created(data)

        elif event_type == "order.paid":

            handle_order_paid(data)

        elif event_type == "order.shipped":

            handle_order_shipped(data)

        else:

            print(f"未知事件类型: {event_type}")



        # 确认消息已被处理（Ack）

        message.ack()

        print(f"消息已确认: {message.message_id}")



    except json.JSONDecodeError as e:

        print(f"消息格式错误: {e}, 内容: {message.data}")

        # 格式错误的消息直接 Ack 掉，避免阻塞队列

        message.ack()

    except Exception as e:

        print(f"处理消息失败: {e}")

        # Nack 消息，允许系统重新投递

        message.nack()





def handle_order_created(data: dict):

    """订单创建后的处理：库存扣减、发货准备"""

    print(f"  → 开始处理订单 #{data['order_id']}")

    print(f"  → 扣减库存: {len(data.get('items', []))} 件商品")

    print(f"  → 验证收货地址: {data.get('shipping_address')}")





def handle_order_paid(data: dict):

    """支付成功的处理：发送确认邮件、更新订单状态"""

    print(f"  → 支付方式: {data.get('payment_method')}")

    print(f"  → 发送支付成功通知给用户 {data.get('user_id')}")





def handle_order_shipped(data: dict):

    """发货后的处理：发送物流信息"""

    print(f"  → 物流单号: {data.get('tracking_number')}")

    print(f"  → 通知用户查看物流信息")





# 启动监听

def listen_for_messages(timeout_seconds: int = None):

    streaming_pull_future = subscriber.subscribe(

        subscription_path, callback=callback

    )

    print(f"开始监听订阅: {subscription_path}")



    try:

        # timeout=None 表示持续监听

        streaming_pull_future.result(timeout=timeout_seconds)

    except TimeoutError:

        streaming_pull_future.cancel()

        streaming_pull_future.result()

    except KeyboardInterrupt:

        streaming_pull_future.cancel()

        streaming_pull_future.result()

        print("监听已停止")





if __name__ == "__main__":

    listen_for_messages()

这段代码有几个设计要点：

幂等性处理：因为 Pub/Sub 是 at-least-once 交付，回调函数应该是幂等的
异常处理：调用
1
message.nack()

让系统重试，而不是 Ack 掉失败的消息
格式校验：JSON 解析失败的消息直接 Ack 并记录日志，不阻塞后续消息
回调是串行的：默认一个 stream 通道内的回调是串行执行的，可以调整流控参数支持并发

消息队列处理流程

实战四：推送订阅（Push Subscription）

拉取模式需要你维护一个长连接进程，而推送模式则允许 Pub/Sub 直接将消息 POST 到你指定的 HTTP 端点。推送订阅非常适合与 Cloud Run、Cloud Functions 或 App Engine 集成。

创建推送订阅


1
2
3
4
5
6
7
8
# 创建推送订阅（指向 Cloud Run 服务）

gcloud pubsub subscriptions create order-events-push-sub \

  --topic=order-events \

  --push-endpoint=https://order-processor-xxx-uc.a.run.app/webhook/pubsub \

  --push-auth-service-account=pubsub-sa@YOUR_PROJECT_ID.iam.gserviceaccount.com \

  --ack-deadline=60 \

  --min-retry-delay=10s \

  --max-retry-delay=600s

使用 Cloud Run 接收推送消息


1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
from flask import Flask, request, jsonify

import json



app = Flask(__name__)



@app.route("/webhook/pubsub", methods=["POST"])

def handle_pubsub_push():

    """

    处理 Pub/Sub 推送消息

    """

    envelope = request.get_json()



    if not envelope:

        msg = "没有收到 Pub/Sub 消息"

        print(msg)

        return jsonify({"error": msg}), 400



    pubsub_message = envelope.get("message", {})

    data_b64 = pubsub_message.get("data", "")



    if not data_b64:

        msg = "消息体中没有 data 字段"

        print(msg)

        return jsonify({"error": msg}), 400



    import base64

    data_bytes = base64.b64decode(data_b64)

    data = json.loads(data_bytes.decode("utf-8"))

    attributes = pubsub_message.get("attributes", {})



    event_type = attributes.get("event_type", "unknown")

    order_id = data.get("order_id", "unknown")

    print(f"[Push] 收到事件: {event_type}, 订单: {order_id}")



    return jsonify({"status": "ok"}), 200



@app.errorhandler(500)

def internal_error(e):

    return jsonify({"error": "Internal server error"}), 500





if __name__ == "__main__":

    app.run(host="0.0.0.0", port=8080)

推送订阅的关键注意事项：

你的端点必须在 60 秒内返回 HTTP 200，否则 Pub/Sub 认为投递失败会重试
建议启用推送认证（
1
--push-auth-service-account

），只有 Pub/Sub 服务可以调用你的端点
推送的请求中包含
1
AUTHORIZATION

Header 用于验证请求来源

实战五：死信队列（Dead Letter Queue）

当消息处理多次失败后，如果不加以处理，这些”坏消息”会不断重试、浪费资源、阻塞正常消息。死信队列（DLQ）就是解决方案：


1
2
3
4
5
6
7
8
# 创建死信主题

gcloud pubsub topics create order-events-dlq



# 在主订阅上配置死信策略

gcloud pubsub subscriptions update order-events-sub \

  --dead-letter-topic=order-events-dlq \

  --max-delivery-attempts=5 \

  --dead-letter-topic-project=$PROJECT_ID

Python SDK 创建带 DLQ 的订阅：


1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
from google.cloud import pubsub_v1



project_id = "your-project-id"

subscriber = pubsub_v1.SubscriberClient()



subscription_path = subscriber.subscription_path(project_id, "order-events-sub")

dlq_topic_path = subscriber.topic_path(project_id, "order-events-dlq")



# 更新订阅配置，添加死信策略

subscriber.update_subscription(

    request={

        "subscription": {

            "name": subscription_path,

            "dead_letter_policy": {

                "dead_letter_topic": dlq_topic_path,

                "max_delivery_attempts": 5,

            },

        },

        "update_mask": {

            "paths": ["dead_letter_policy"],

        },

    }

)

print("死信策略已配置: 最多重试 5 次")

死信队列最佳实践：

为死信主题创建独立的订阅，专门分析和处理异常消息
设置合理的最多重试次数（通常 5-10 次），既给临时故障恢复机会，又不浪费太多资源
死信消息处理逻辑应记录详细的失败上下文，便于排查问题
定期巡检死信队列，避免积压过多未处理的消息

实战六：消息过滤与顺序保证

过滤订阅（Filter Subscription）

如果某个消费者只关心特定类型的事件，可以使用过滤订阅减少消息量和网络开销：


1
2
3
4
5
6
7
8
9
# 创建只接收订单支付事件的过滤订阅

gcloud pubsub subscriptions create order-paid-only-sub \

  --topic=order-events \

  --filter='attributes.event_type = "order.paid"'



# 创建接收金额大于 500 的大额订单事件

gcloud pubsub subscriptions create large-order-sub \

  --topic=order-events \

  --filter='attributes.event_type = "order.created" AND attributes.amount > 500'

过滤表达式支持常见的比较操作：

、

（子串匹配）、

AND

、

NOT

、

has

（标签存在性检查）。

消息顺序保证

Pub/Sub 默认不保证消息顺序，但如果你使用 ordering key，同一 key 的消息会被按发布顺序投递：


1
2
3
4
5
6
7
8
9
10
11
# 创建启用 ordering key 的订阅

gcloud pubsub subscriptions create order-events-ordered-sub \

  --topic=order-events \

  --enable-message-ordering



# 发布时指定 ordering key

future = publisher.publish(

    topic_path,

    data=data_bytes,

    ordering_key=order_data["order_id"],

)

注意使用 ordering key 会带来一些性能开销（约 10-20% 的吞吐量下降），只在确实需要顺序保证的场景下使用。

生产环境最佳实践

流控与并发


1
2
3
4
5
6
7
8
9
10
11
12
13
14
# 控制并发流，防止消费者被压垮

subscriber = pubsub_v1.SubscriberClient()



# 设置流控参数

flow_control = pubsub_v1.types.FlowControl(

    max_messages=100,

    max_bytes=10 * 1024 * 1024,

)



streaming_pull_future = subscriber.subscribe(

    subscription_path,

    callback=callback,

    flow_control=flow_control,

)

监控与告警

Pub/Sub 在 Cloud Monitoring 中提供了丰富的指标：

指标名称

说明

建议告警阈值

1	subscriber/num_undelivered_messages

未确认的消息数

> 10,000 持续 5 分钟

1	subscription/dead_letter_message_count

死信消息数

> 100

1	subscription/oldest_unacked_message_age

最久未 Ack 消息的存量时长

> 30 分钟

成本控制

Pub/Sub 的计费主要包括三部分：

数据吞吐量：按 GB 计费，每 GB 约 $0.04
请求次数：每 64KB 算一次请求
存储费用：消息存储按 GB/月计费

优化建议：

使用 批量发布（Batching），将多条小消息合并为一次请求
充分利用 过滤订阅，减少不必要的数据传输
设置合理的 消息保留期限

安全最佳实践

始终使用服务账号而非用户账号进行认证
为每个服务创建最小权限的服务账号
推送订阅务必启用
1
push-auth

，验证请求来源
敏感数据传输时启用 CMEK 或 CSEK
使用 VPC Service Controls 防止数据外泄

总结

Google Cloud Pub/Sub 是一个功能丰富、性能出色的全托管消息队列服务。通过本文的实战演练，我们掌握了：

Pub/Sub 的核心概念与架构设计
使用 gcloud CLI 和 Python SDK 创建主题与订阅
实现消息发布与订阅（包括推送模式和拉取模式）
配置死信队列处理异常消息
使用过滤订阅和 ordering key 实现精细化控制
生产环境的流控、监控、成本优化和安全加固方案

无论你是正在建设微服务架构的后端工程师，还是需要设计事件驱动系统的架构师，Pub/Sub 都值得认真考虑。它不仅是 Google Cloud 生态系统中的核心组件，其设计理念也对整个消息中间件领域产生了深远影响。

建议你动手创建一个 GCP 项目，跟着本文的代码一步步搭建自己的第一个 Pub/Sub 消息系统——纸上学来终觉浅，绝知此事要躬行。