如何使用Python可靠拉取Google Pub/Sub消息？pull返回空消息问题咨询

你的代码确实存在问题，这是Google Pub/Sub同步拉取模式的典型误用场景，核心问题和修复方案如下：

• Google Pub/Sub 是分布式架构，消息分散存储在多个集群节点上，单次同步pull请求只会请求部分节点，若请求的节点暂时没有可投递的消息（消息存在其他节点、或还在写入缓存未完成同步），就会返回空响应，这是正常设计，不代表全局没有待处理消息。你的代码在第一次拿到空响应就直接break退出循环，必然会漏掉大量积压消息。
• 你未配置pull请求的超时参数，默认情况下pull请求不会等待消息到达，短时间内没有可分发的消息就会直接返回空，进一步拉高了空响应的出现概率。
• 代码未做异常捕获，单条消息格式异常、拉取请求超时等问题都会直接导致程序终止，未处理的消息会重新入队，但也可能被误判为队列已空。

修改后的代码如下，核心优化点：移除单次空响应直接退出的逻辑，增加空响应重试阈值，添加超时和异常捕获：

import os
import time
from google.cloud import pubsub
import ast

PROJECT_ID = os.environ['PROJECT_ID']
subscriber = pubsub.SubscriberClient()
subscription_path = subscriber.subscription_path(PROJECT_ID, 'subscription-name')

# 连续空响应阈值，超过该值再判定队列确实无积压
MAX_EMPTY_RETRIES = 3
empty_retry_count = 0

while empty_retry_count < MAX_EMPTY_RETRIES:
    try:
        response = subscriber.pull(
            request={
                "subscription": subscription_path,
                "max_messages": 50,
            },
            # 设置10秒超时，有消息就立刻返回，最多等10秒再返回，降低空响应概率
            timeout=10
        )
    except Exception as e:
        print(f"拉取消息异常，1秒后重试: {str(e)}")
        time.sleep(1)
        continue

    if not response.received_messages:
        empty_retry_count += 1
        print(f'第{empty_retry_count}次未拉取到消息，2秒后重试...')
        time.sleep(2)
        continue
    
    # 拉取到消息重置空重试计数
    empty_retry_count = 0
    ack_ids = []
    for msg in response.received_messages:
        try:
            message_data = ast.literal_eval(msg.message.data.decode('utf-8'))
            # 此处保留原有数据转换、投递到其他主题的逻辑
            ack_ids.append(msg.ack_id)
        except Exception as e:
            print(f"消息处理失败，消息ID: {msg.message.message_id}, 错误信息: {str(e)}")
            # 处理失败的消息不ack，后续会自动重新投递

    if ack_ids:
        subscriber.acknowledge(
            request={
                "subscription": subscription_path,
                "ack_ids": ack_ids,
            }
        )

print('🏁 队列已无待处理消息，程序退出...')

• 如果是长期运行的消费场景，更推荐使用Pub/Sub官方提供的异步消费模式，通过subscriber.subscribe()方法注册回调函数，客户端会自动处理拉取、重试、流控逻辑，稳定性远高于自行实现的同步拉取逻辑。
• 如果需要明确确认队列是否真的没有积压，可以调用云监控API查询对应订阅的num_undelivered_messages指标，不要完全依赖拉取请求的返回结果判断积压状态。

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