Python Pub/Sub 订阅者客户端在使用过滤器时无法拉取消息的解决方案

在使用 Google Cloud Pub/Sub 时，一个常见的问题是，当订阅配置了消息过滤器后，订阅者客户端无法拉取消息。即使手动在控制台触发“Pull”操作，也能看到符合过滤条件的消息存在于订阅中，但客户端仍然无法接收。以下详细分析了这个问题的原因和解决方案。

问题分析

问题的根本原因在于，当使用过滤器创建订阅后，过滤器可能需要一些时间才能完全生效并正确注册订阅者客户端。如果订阅者客户端在过滤器生效之前立即创建，则可能会导致客户端无法正确接收消息。

解决方案

解决方案的核心在于确保订阅者客户端在过滤器完全生效后才开始拉取消息。以下是几种可行的方案：

1. 延迟启动订阅者客户端: 在创建订阅后，添加一个短暂的延迟（例如，几秒钟的睡眠时间），以确保过滤器有足够的时间生效。

   立即学习“Python免费学习笔记（深入）”；

   

   Veo

   Google 最新发布的 AI 视频生成模型

   下载

   import time
   import os
   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

2. 重试机制: 如果在启动客户端后仍然无法拉取消息，可以实现一个重试机制，定期检查订阅是否准备就绪，并尝试重新启动订阅者客户端。

3. 健康检查: 实施健康检查机制，定期验证订阅者客户端是否能够成功拉取消息。如果健康检查失败，则重新启动客户端。

注意事项

• 延迟时间的长短取决于具体的环境和配置。建议根据实际情况进行调整。
• 在生产环境中，建议使用更健壮的解决方案，例如重试机制或健康检查，以确保订阅者客户端的稳定性和可靠性。
• 确保订阅的过滤器配置正确，并且符合预期的消息过滤规则。

总结

当 Pub/Sub 订阅配置了过滤器时，确保订阅者客户端在过滤器完全生效后才开始拉取消息至关重要。通过添加延迟、实现重试机制或实施健康检查，可以有效地解决客户端无法拉取消息的问题，确保应用程序能够正确接收和处理消息。