如何用Cloud Functions消费Pub/Sub消息并避免未处理消息引发无限执行？

1. 配置重试阈值与死信队列 #

给Pub/Sub订阅设置明确的重试规则，超过阈值的失败消息自动转入死信队列（DLQ），彻底终止无效重试：

• 设置maxDeliveryAttempts（最大投递次数），比如5次，超过后消息不再投递
• 设置messageRetentionDuration（消息留存时长），比如1天，避免旧消息长期占用资源
• 关联死信主题，后续可单独分析死信队列里的消息，排查失败原因

2. 实现幂等处理逻辑 #

确保同一条消息多次处理不会产生重复副作用：

• 用Pub/Sub消息的event_id作为唯一标识，处理前先在数据库或缓存（如Redis）中检查是否已处理过该消息
• 处理成功后标记该消息ID为已处理，后续重试时直接跳过

示例代码（Python）：

import redis

redis_client = redis.Redis()

def process_message(event, context):
    msg_id = context.event_id
    # 检查消息是否已处理
    if redis_client.get(msg_id):
        return "Skip processed message"
    
    try:
        # 执行核心业务逻辑
        handle_business_logic(event)
        # 标记消息为已处理
        redis_client.setex(msg_id, 86400, "processed")
    except Exception as e:
        # 抛出异常触发重试（未达重试阈值时）
        raise e

3. 区分错误类型控制重试 #

针对不同错误类型决定是否触发重试：

• 临时错误（如网络波动、服务暂时不可用）：抛出异常，让Pub/Sub自动重试
• 永久错误（如消息格式错误、参数非法）：不抛出异常，直接确认消息，终止重试

示例代码（Python）：

class TemporaryError(Exception):
    pass

class PermanentError(Exception):
    pass

def handle_business_logic(event):
    msg_data = event.get("data")
    if not msg_data:
        # 消息格式错误，永久错误
        raise PermanentError("Invalid message format")
    
    try:
        # 调用外部服务
        call_external_service(msg_data)
    except ConnectionError:
        # 网络问题，临时错误
        raise TemporaryError("Service unavailable")

4. 优化批量处理设置 #

如果单条处理效率低，可开启Cloud Function的批量处理功能，减少函数重复启动：

• 设置maxMessagesPerBatch（每批最大消息数）和batchSize（批量处理大小）
• 批量处理时对每条消息单独处理，避免一条失败导致整批消息重复投递

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