如何在长耗时任务处理时持续更新Cloud Pub/Sub的ack确认截止时间

无需使用复杂的asyncio逻辑，通过线程实现后台续期更简单，且完全兼容你现有的同步处理代码，仅需给回调函数加一个装饰器即可完成需求。

完整代码实现 #

import threading
import time
from google.cloud import pubsub_v1

# 续期装饰器，可自定义续期间隔和每次延长的时长
def auto_extend_ack_deadline(extend_interval: int = 590, extend_seconds: int = 600):
    def decorator(processing_func):
        def wrapper(message):
            # 处理完成的信号标记
            processing_done = threading.Event()
            
            # 后台续期逻辑
            def extend_worker():
                # 每到间隔时间就检查是否处理完成，未完成则续期
                while not processing_done.wait(extend_interval):
                    message.modify_ack_deadline(seconds=extend_seconds)
            
            # 启动后台续期线程，daemon设为True避免主线程退出时卡住
            extend_thread = threading.Thread(target=extend_worker, daemon=True)
            extend_thread.start()
            
            try:
                # 执行原有业务处理逻辑
                return processing_func(message)
            except Exception as e:
                # 业务逻辑报错时主动nack，让消息可以重新投递
                message.nack()
                raise e
            finally:
                # 无论处理成功还是失败，都标记完成，停止续期线程
                processing_done.set()
        return wrapper
    return decorator

# -------------------------- 原有业务代码修改部分 --------------------------
subscriber = pubsub_v1.SubscriberClient()
subscription_path = "projects/path/to/subscription"

# 仅需加这一行装饰器即可实现自动续期
@auto_extend_ack_deadline()
def callback(message):
    print(f'Received message: {message}')
    print(f'data: {message.data}')

    if message.attributes:
        # 此处为长耗时处理逻辑
        # ...

    message.ack()

# subscribe方法提供了异步处理回调的接口
streaming_pull_future = subscriber.subscribe(subscription_path, callback=callback)
with subscriber:                                           # 使用with块包裹subscriber，执行完毕后自动调用close()
    try:
        streaming_pull_future.result()                          # 不设置超时将无限阻塞等待
    except TimeoutError:
        streaming_pull_future.cancel()                          # 触发关闭流程
        streaming_pull_future.result()

关键逻辑说明 #

• 用threading.Event作为处理完成的信号，无需额外加锁，性能开销极低
• 后台线程默认每590秒触发一次续期，每次将ack截止时间重置为600秒，只要处理未完成就会持续续期
• 无论业务逻辑执行成功还是抛出异常，都会先触发完成信号停止续期线程，不会出现线程泄漏
• 业务报错时自动调用nack把消息放回队列，避免消息丢失

注意事项 #

• 续期间隔建议比延长时间少10~30秒，预留网络请求的耗时，避免续期请求还没完成就到了原来的截止时间
• 建议根据业务最大处理时长添加最大续期次数限制，避免异常卡住的消息被永远续期，无法投递到其他消费者
• 如果需要用asyncio实现，只需要把线程替换为asyncio任务、time.sleep替换为asyncio.sleep、信号用asyncio.Event即可，核心逻辑完全一致

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