Python Asyncio run_forever()与Tasks：Google Cloud PubSub异步代码咨询

作为经常处理异步PubSub场景的开发者，我来给你梳理下这类场景下的核心问题和解决方案，结合你给出的代码片段展开：

1. 如何在PubSub异步生产/消费中正确使用run_forever()？ #

run_forever()的核心作用是让事件循环持续运行，直到你主动调用loop.stop()——这刚好匹配PubSub消费者/生产者长期运行的需求。咱们结合你的代码，调整出一个可运行的完整示例：

import asyncio
import datetime
import os
from google.cloud import pubsub

async def message_producer(publisher_client, topic_path):
    """Publish messages with current datetime"""
    while True:
        msg_content = datetime.datetime.now().isoformat().encode("utf-8")
        await publisher_client.publish(topic_path, msg_content)
        await asyncio.sleep(0.1)

async def proc_message(message):
    await asyncio.sleep(0.1)
    print(f"Processed message: {message.data.decode('utf-8')}")
    message.ack()

async def message_consumer(subscriber_client, subscription_path):
    """Listen for PubSub messages indefinitely"""
    flow_control = pubsub.types.FlowControl(max_messages=10)
    async with subscriber_client.subscribe(
        subscription_path, flow_control=flow_control
    ) as subscriber:
        async for message in subscriber:
            await proc_message(message)

def main():
    # 初始化PubSub异步客户端
    publisher = pubsub.PublisherClient()
    subscriber = pubsub.SubscriberClient()
    
    project_id = os.getenv("GOOGLE_CLOUD_PROJECT")
    topic_path = publisher.topic_path(project_id, "test-topic")
    sub_path = subscriber.subscription_path(project_id, "test-sub")
    
    # 获取事件循环实例
    loop = asyncio.get_event_loop()
    
    # 将协程包装成Task，加入事件循环调度队列
    producer_task = loop.create_task(message_producer(publisher, topic_path))
    consumer_task = loop.create_task(message_consumer(subscriber, sub_path))
    
    try:
        # 启动事件循环，持续运行直到手动停止
        loop.run_forever()
    except KeyboardInterrupt:
        # 捕获中断信号，优雅取消所有任务
        producer_task.cancel()
        consumer_task.cancel()
        # 等待任务完成清理逻辑
        loop.run_until_complete(
            asyncio.gather(producer_task, consumer_task, return_exceptions=True)
        )
    finally:
        # 关闭事件循环，释放资源
        loop.close()

if __name__ == "__main__":
    main()

这里的关键要点：

• 必须用loop.create_task()（或Python3.7+的asyncio.create_task()）把协程包装成Task，事件循环才会调度它们执行
• run_forever()会阻塞当前线程，直到你主动停止循环，完美适配PubSub长期运行的场景

2. 常见误区：为什么我的Task没执行？ #

• 忘记将协程加入事件循环：只定义协程是没用的，必须把它包装成Task并加入事件循环的任务队列，否则run_forever()只会空转
• 误用run_until_complete()：如果用这个方法替代run_forever()，当传入的协程完成后循环就会停止——而你的生产者/消费者是无限循环的，所以必然会导致服务退出
• 协程内存在同步阻塞调用：如果在异步协程里调用了同步的PubSub方法（比如非async版本的publish），会直接卡住整个事件循环，导致其他Task无法被调度

3. 如何优雅停止run_forever()和Tasks？ #

• 优先通过捕获信号（比如KeyboardInterrupt）触发停止逻辑，避免强制杀死进程
• 取消Task时，用asyncio.gather(..., return_exceptions=True)可以避免取消操作抛出未捕获的异常
• 关闭事件循环前，确保PubSub客户端等资源被正确释放（示例中用async with管理消费者连接，会自动处理）

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