本文档旨在解决 Python Pub/Sub 订阅者客户端在应用订阅过滤器后无法拉取消息的问题。通过分析问题原因，提供了一种简单有效的解决方案，即在创建订阅后添加短暂的延迟，确保订阅完全生效后再创建订阅者客户端。
问题分析
在使用 Google Cloud Pub/Sub 的 Python 客户端时，有时会遇到这样的情况：当未对订阅应用任何过滤器时，订阅者客户端能够成功地从订阅中拉取消息；然而，一旦应用了订阅过滤器，客户端就无法再拉取消息。尽管手动在 Pub/Sub 控制台中拉取消息可以正常工作，表明消息确实符合过滤器条件并存在于订阅中，但客户端却无法接收它们。
原因探究
根本原因在于，当使用过滤器创建订阅时，可能需要一些时间才能使订阅完全生效并正确注册订阅者客户端。如果在订阅创建后立即创建订阅者客户端，客户端可能无法正确地与已过滤的订阅关联，导致无法拉取消息。
解决方案
解决此问题的最简单有效的方法是在创建订阅后添加一个短暂的延迟，然后再创建订阅者客户端。这可以通过在代码中添加 time.sleep() 函数来实现。
立即学习“Python免费学习笔记（深入）”；
以下是一个示例代码片段，演示了如何添加延迟：
import timefrom 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 osimport timefrom google.cloud import pubsub_v1from app.services.subscription_service import save_bill_eventsfrom app.utils.constants import BILL_SUBSCRIPTION_GCP_PROJECT_ID, BILL_EVENT_SUBSCRIPTION_IDfrom app.utils.logging_tracing_manager import get_loggerlogger = 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 服务的响应速度调整延迟时间，以达到最佳效果。
以上就是Python Pub/Sub 订阅者客户端在应用过滤器后无法拉取消息的解决方案的详细内容，更多请关注创想鸟其它相关文章！
版权声明：本文内容由互联网用户自发贡献，该文观点仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。
如发现本站有涉嫌抄袭侵权/违法违规的内容， 请发送邮件至 chuangxiangniao@163.com 举报，一经查实，本站将立刻删除。
发布者：程序猿，转转请注明出处：https://www.chuangxiangniao.com/p/1362947.html
微信扫一扫 
支付宝扫一扫 
