本文档旨在解决 Python Pub/Sub 订阅者客户端在应用订阅过滤器后无法拉取消息的问题。通过分析问题原因，提供了一种简单有效的解决方案，即在创建订阅后添加短暂的延迟，确保订阅完全生效后再创建订阅者客户端。
问题分析
在使用 Google Cloud Pub/Sub 的 Python 客户端时，有时会遇到这样的情况：当未对订阅应用任何过滤器时，订阅者客户端能够成功地从订阅中拉取消息；然而，一旦应用了订阅过滤器，客户端就无法再拉取消息。尽管手动在 Pub/Sub 控制台中拉取消息可以正常工作，表明消息确实符合过滤器条件并存在于订阅中，但客户端却无法接收它们。
原因探究
根本原因在于，当使用过滤器创建订阅时，可能需要一些时间才能使订阅完全生效并正确注册订阅者客户端。如果在订阅创建后立即创建订阅者客户端，客户端可能无法正确地与已过滤的订阅关联，导致无法拉取消息。
解决方案
解决此问题的最简单有效的方法是在创建订阅后添加一个短暂的延迟，然后再创建订阅者客户端。这可以通过在代码中添加 time.sleep() 函数来实现。
立即学习“Python免费学习笔记（深入）”；
以下是一个示例代码片段，演示了如何添加延迟：
import time
from google.cloud import pubsub_v1
# 创建订阅 (假设 subscription_path 已经定义)
# subscriber.create_subscription(request={"name": subscription_path, "topic": topic_path, "filter": "attributes.key = 'value'"})
# 添加延迟，例如 5 秒
time.sleep(5)
# 创建订阅者客户端
subscriber = pubsub_v1.SubscriberClient()
# ... 订阅和消息处理逻辑 ...解释：
- time.sleep(5)：这行代码会暂停程序的执行 5 秒钟。您可以根据实际情况调整延迟时间。
完整代码示例
下面是一个完整的示例，展示了如何在订阅创建后添加延迟，以确保订阅者客户端能够正常拉取消息：
import os
import time
from google.cloud import pubsub_v1
from app.services.subscription_service import save_bill_events
from app.utils.constants import BILL_SUBSCRIPTION_GCP_PROJECT_ID, BILL_EVENT_SUBSCRIPTION_ID
from app.utils.logging_tracing_manager import get_logger
logger = get_logger(__file__)
def callback(message: pubsub_v1.subscriber.message.Message) -> None:
save_bill_events(message.data)
message.ack()
# 在创建订阅后添加延迟
# 假设以下代码在创建订阅之后执行
time.sleep(5)
subscriber = pubsub_v1.SubscriberClient()
subscription_path = subscriber.subscription_path(os.environ.get(BILL_SUBSCRIPTION_GCP_PROJECT_ID),
BILL_EVENT_SUBSCRIPTION_ID)
# Limit the subscriber to only have fixed number of outstanding messages at a time.
flow_control = pubsub_v1.types.FlowControl(max_messages=50)
streaming_pull_future = subscriber.subscribe(subscription_path, callback=callback, flow_control=flow_control)
async def poll_bill_subscription():
with subscriber:
try:
# When `timeout` is not set, result() will block indefinitely,
# unless an exception is encountered first.
streaming_pull_future.result()
except Exception as e:
# Even in case of an exception, subscriber should keep listening
logger.error(
f"An error occurred while pulling message from subscription {BILL_EVENT_SUBSCRIPTION_ID}",
exc_info=True)
pass注意事项：
- 延迟时间的选择取决于您的具体环境和 Pub/Sub 服务的响应速度。建议根据实际情况进行调整。
- 在某些情况下，可能需要更长的延迟时间才能确保订阅完全生效。
- 除了添加延迟之外，您还可以尝试在创建订阅者客户端之前显式地检查订阅的状态，以确保订阅已成功创建并已启用过滤器。
总结
当 Python Pub/Sub 订阅者客户端在应用订阅过滤器后无法拉取消息时，通常是因为订阅尚未完全生效。通过在创建订阅后添加一个短暂的延迟，可以确保订阅者客户端能够正确地与已过滤的订阅关联，从而解决此问题。这种方法简单有效，可以避免不必要的错误和麻烦。在实际应用中，请根据您的具体环境和 Pub/Sub 服务的响应速度调整延迟时间，以达到最佳效果。
