使用GCP Pub/Sub拉取订阅时无法获取delivery_attempts字段的问题

核心原因 #

delivery_attempts 字段的返回确实和死信队列(DLQ)策略直接相关——GCP Pub/Sub仅在订阅配置了死信策略时，才会自动追踪并返回消息的投递次数。不管是同步拉取的 pubsub_v1.types.PubsubMessage 还是异步拉取的 pubsub_v1.subscriber.message.Message，没有DLQ策略的话，要么字段不存在，要么返回null，这是服务端的设计逻辑，和SDK版本无关。

不配置死信策略的同步模式解决方案 #

如果你不想启用死信策略，只能通过自定义逻辑来追踪投递次数，以下是两种可行方案：

1. 外部存储维护投递计数

• 利用消息的唯一标识 message_id，借助Redis、Cloud Datastore等外部存储来记录每条消息的投递次数：
  • 同步拉取到消息后，先查询存储中该 message_id 的计数：
    • 若不存在，初始化计数为1；
    • 若已存在，将计数递增1。
  • 当消息处理成功并确认（ack()）后，可删除存储中的对应记录，避免重复计数；若处理失败需要重试，保留计数以便下次更新。
  • 示例代码片段（用Redis）：

    import redis
    from google.cloud import pubsub_v1
    
    redis_client = redis.Redis(host='your-redis-host', port=6379, db=0)
    subscriber = pubsub_v1.SubscriberClient()
    subscription_path = subscriber.subscription_path('project-id', 'subscription-name')
    
    def sync_pull_messages():
        response = subscriber.pull(subscription_path, max_messages=10)
        for msg in response.received_messages:
            msg_id = msg.message.message_id
            # 获取当前投递次数
            attempt_count = redis_client.incr(msg_id)
            print(f"Message {msg_id} delivery attempt: {attempt_count}")
            
            # 处理消息逻辑...
            # 处理成功后确认并删除计数
            subscriber.acknowledge(subscription_path, [msg.ack_id])
            redis_client.delete(msg_id)
    

2. 生产者自定义投递属性（仅适用于可控生产者场景）

• 如果消息是由你自己的服务生产的，在发送消息时添加自定义属性（比如 x-delivery-attempt），初始值设为1：

  from google.cloud import pubsub_v1
  
  publisher = pubsub_v1.PublisherClient()
  topic_path = publisher.topic_path('project-id', 'topic-name')
  
  data = b"your-message-data"
  # 添加自定义投递次数属性
  attributes = {"x-delivery-attempt": "1"}
  future = publisher.publish(topic_path, data, **attributes)
  

• 当消费者处理失败需要重试时，递增该属性值后重新发布消息（如果是通过重新发布实现重试的话），这样同步拉取时就能从消息属性中获取投递次数。

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