GCP 云函数开发指南：Python + HTTP 触发 + Pub/Sub

以下是完整的实现方案，包含开发步骤、核心代码和部署说明：

1. 核心功能设计

• HTTP 触发器：接收 POST/GET 请求
• Pub/Sub 集成：将请求数据发布到指定主题
• 数据流：HTTP 请求 → 云函数处理 → Pub/Sub 消息发布

2. 开发环境准备

# 安装必要库
pip install functions-framework google-cloud-pubsub flask

3. 函数代码 (main.py)

import json
from flask import jsonify, request
from google.cloud import pubsub_v1

# 初始化 Pub/Sub 客户端
publisher = pubsub_v1.PublisherClient()
TOPIC_PATH = 'projects/YOUR_PROJECT_ID/topics/YOUR_TOPIC_NAME'  # 替换实际值

def publish_message(data):
    """发布消息到 Pub/Sub 主题"""
    data_str = json.dumps(data).encode('utf-8')
    future = publisher.publish(TOPIC_PATH, data_str)
    return future.result()  # 返回消息ID

def http_trigger(request):
    """HTTP 触发的入口函数"""
    # 1. 验证请求方法
    if request.method != 'POST':
        return jsonify({"error": "只支持 POST 请求"}), 405
    
    # 2. 解析请求数据
    try:
        payload = request.get_json()
        if not payload:
            raise ValueError("空请求体")
    except Exception as e:
        return jsonify({"error": f"数据解析失败: {str(e)}"}), 400
    
    # 3. 发布到 Pub/Sub
    try:
        msg_id = publish_message(payload)
        return jsonify({
            "status": "消息已发布",
            "message_id": msg_id,
            "topic": TOPIC_PATH
        }), 200
    except Exception as e:
        return jsonify({"error": f"发布失败: {str(e)}"}), 500

4. 本地测试脚本 (test_script.py)

import requests

URL = "http://localhost:8080"  # 本地测试地址
payload = {
    "user_id": "cloud_user_123",
    "action": "file_upload",
    "timestamp": "2023-10-05T08:30:00Z"
}

response = requests.post(URL, json=payload)
print("状态码:", response.status_code)
print("响应内容:", response.json())

5. 部署到 GCP

步骤：

1. 创建 requirements.txt：

   functions-framework==3.*
   google-cloud-pubsub==2.13.0
   flask==2.2.2
   

2. 使用 gcloud 部署：

   gcloud functions deploy http-pubsub-bridge \
     --runtime python310 \
     --trigger-http \
     --allow-unauthenticated \
     --region=asia-east1 \
     --entry-point=http_trigger
   

6. 关键配置说明

7. 验证流程

1. 发送测试请求：

   curl -X POST https://REGION-PROJECT_ID.cloudfunctions.net/http-pubsub-bridge \
     -H "Content-Type: application/json" \
     -d '{"test": "pubsub_integration"}'
   

2. 预期响应：

   {
     "status": "消息已发布",
     "message_id": "1234567890",
     "topic": "projects/my-project/topics/data-ingest"
   }
   

3. 在 GCP Pub/Sub 控制台 订阅主题验证消息接收

8. 安全增强建议

1. 添加身份验证：

   gcloud functions remove-invoker-policy-binding http-pubsub-bridge \
     --member="allUsers"
   

2. 在函数开头添加 JWT 验证：

   from google.auth import jwt, transport
   def verify_token(request):
       header = request.headers.get("Authorization")
       # 实现 JWT 验证逻辑
   

注意：实际部署时需替换 YOUR_PROJECT_ID 和 YOUR_TOPIC_NAME，生产环境务必配置 VPC 和服务账号权限。