如何从Pub/Sub触发的Python Cloud Function中获取delivery_attempt属性

前置条件 #

你首先需要给Cloud Function触发绑定的Pub/Sub订阅配置死信主题（DLQ），同时设置最大消息传递次数。只有配置了死信队列的订阅，才会向Cloud Function传递delivery_attempt字段，未配置的场景下该字段始终为空。

获取代码 #

你可以直接从context参数中提取该字段，示例代码如下：

def hello_pubsub(event, context):
    """Background Cloud Function to be triggered by Pub/Sub.
    Args:
         event (dict):  Pub/Sub消息主体，包含data、attributes等字段
         context (google.cloud.functions.Context): 触发事件元数据
    """
    # 提取delivery_attempt，未配置DLQ时默认返回None
    delivery_attempt = context.delivery_attempt
    
    # 业务逻辑示例：如果重试超过3次就直接抛弃，不执行后续逻辑
    if delivery_attempt and delivery_attempt > 3:
        print(f"消息已重试{delivery_attempt}次，超过阈值直接跳过")
        return
    
    # 剩下的业务逻辑

注意事项 #

• delivery_attempt从1开始计数，第一次触发函数时值为1，每重试一次数值加1
• 当传递次数达到你在DLQ配置的最大次数后，消息会被自动转发到死信主题，不会再触发函数
• 如果需要兼容旧版本运行时，也可以用如下方式提取，效果一致：

  delivery_attempt = getattr(context, 'delivery_attempt', None)
  

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