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

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()

# ... 订阅和消息处理逻辑 ...

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