如何使用AWS Lambda消费GCP Pub/Sub主题消息？求实现示例

前置准备 #

• AWS账号及Lambda服务操作权限
• GCP账号及Pub/Sub管理权限
• 创建GCP服务账号，分配roles/pubsub.subscriber权限，生成并下载JSON格式的密钥文件

步骤1：GCP侧配置 #

1. 若目标主题不存在，先创建Pub/Sub主题
2. 为该主题创建拉取式订阅（必须用拉取模式，Lambda无法直接接收Pub/Sub推送）
3. 将GCP服务账号密钥上传至AWS Secrets Manager，记录Secret的名称或ARN

步骤2：AWS Lambda配置 #

2.1 配置Lambda执行角色 #

给Lambda执行角色添加以下权限：

• secretsmanager:GetSecretValue：用于读取存储的GCP密钥
• logs:CreateLogGroup、logs:CreateLogStream、logs:PutLogEvents：用于日志输出

2.2 编写Lambda代码（Python示例） #

先安装依赖包：

pip install google-cloud-pubsub -t .

编写核心代码：

import os
import json
import boto3
from google.cloud import pubsub_v1
from google.oauth2 import service_account

def lambda_handler(event, context):
    # 从环境变量读取配置
    SUBSCRIPTION_NAME = os.environ.get('PUBSUB_SUBSCRIPTION_NAME')
    SECRET_NAME = os.environ.get('GCP_SECRET_NAME')
    
    # 从Secrets Manager获取GCP密钥
    secrets_manager = boto3.client('secretsmanager')
    secret_response = secrets_manager.get_secret_value(SecretId=SECRET_NAME)
    gcp_credentials = json.loads(secret_response['SecretString'])
    
    # 初始化Pub/Sub订阅客户端
    credentials = service_account.Credentials.from_service_account_info(gcp_credentials)
    subscriber = pubsub_v1.SubscriberClient(credentials=credentials)
    
    # 批量拉取消息（最多10条，可按需调整）
    response = subscriber.pull(
        request={"subscription": SUBSCRIPTION_NAME, "max_messages": 10}
    )
    
    ack_ids = []
    for received_message in response.received_messages:
        # 打印消息内容，替换为你的业务处理逻辑
        print(f"Received message: {received_message.message.data.decode('utf-8')}")
        # process_message(received_message.message.data)
        
        ack_ids.append(received_message.ack_id)
    
    # 确认消息已处理，避免重复消费
    if ack_ids:
        subscriber.acknowledge(
            request={"subscription": SUBSCRIPTION_NAME, "ack_ids": ack_ids}
        )
    
    return {
        'statusCode': 200,
        'body': json.dumps(f"Processed {len(ack_ids)} messages")
    }

2.3 打包部署Lambda #

将代码和依赖包一起打包成ZIP文件，上传至Lambda函数：

• 配置环境变量：PUBSUB_SUBSCRIPTION_NAME（GCP订阅完整路径，格式为projects/[GCP_PROJECT_ID]/subscriptions/[SUBSCRIPTION_NAME]）、GCP_SECRET_NAME（Secrets Manager中存储密钥的名称）
• 设置Lambda超时时间（建议30秒以上，根据消息处理耗时调整）

步骤3：触发Lambda拉取消息 #

由于Pub/Sub拉取模式需要主动发起请求，可通过以下方式触发Lambda：

• CloudWatch Events定时触发：设置固定间隔（如1分钟）自动运行Lambda拉取消息
• EventBridge调度：支持更灵活的定时规则（如特定时段执行）
• API Gateway触发：手动或外部系统调用触发拉取

注意事项 #

• 消息处理失败时，不要调用acknowledge，Pub/Sub会自动将消息重新放回订阅等待重试
• 若需处理大量消息，可调整max_messages参数，同时匹配Lambda的内存和超时配置
• 定期轮换GCP服务账号密钥，降低泄露风险

内容的提问来源于stack exchange，提问作者jrsall92