跨文件使用Asyncio循环引用时Future函数未执行问题排查

从你的描述来看，你碰到的核心问题是跨脚本传递事件循环和Future任务时，上下文不匹配导致任务没法被正确调度。结合Google Pub/Sub的异步特性和Python asyncio的工作机制，我整理了几个常见原因和对应的修复方案：

1. 先确保事件循环是同一个 #

Pub/Sub的订阅者回调默认是同步执行的——也就是说，回调函数是在订阅者的工作线程里跑的，如果你直接把当前的loop传到initEvga里，很可能出现：这个loop和initEvga所在的异步上下文根本不是同一个，那Future自然没法被监听调度。

怎么改？
要么把回调改成异步的（用async def定义），让回调本身就跑在主事件循环里；要么在同步回调里，手动把异步任务提交到正确的事件循环中。

给你个异步回调的示例：

# 改成异步回调，这样就和主事件循环在同一个上下文里了
async def callback(message):
    try:
        # 直接拿当前活跃的事件循环，不用手动传也可以
        current_loop = asyncio.get_running_loop()
        await initEvga(message, current_loop)
        message.ack()
    except Exception as e:
        print(f"处理消息炸了: {e}")
        message.nack()

# 初始化订阅者的时候直接用这个异步回调
subscriber = pubsub_v1.SubscriberClient()
subscription_path = subscriber.subscription_path(config.PROJECT_ID, config.SUBSCRIPTION_ID)
streaming_pull_future = subscriber.subscribe(subscription_path, callback=callback)

2. 检查Future的创建和触发逻辑 #

如果initEvga里的Future是随便创建的（比如直接asyncio.Future()），那它会绑定到创建时的事件循环上——要是你后来在另一个loop里去触发它，肯定没用。

关键要注意这两点：

• 创建Future的时候，一定要用你传递过来的loop来创建：loop.create_future()，而不是直接asyncio.Future()。
• 触发Future的时候（调用set_result），如果是在同步线程里操作，必须用loop.call_soon_threadsafe()来确保操作跑在正确的loop线程中。

举个initEvga里的正确写法：

async def initEvga(message, loop):
    # 用传过来的loop创建Future，绑定正确的上下文
    my_future = loop.create_future()
    
    # 模拟一个需要触发Future的同步任务（比如调用某个阻塞的API）
    def trigger_future():
        # 解析Pub/Sub消息
        msg_data = json.loads(base64.b64decode(message.data).decode())
        # 确保在目标loop的线程里设置Future结果
        loop.call_soon_threadsafe(my_future.set_result, msg_data)
    
    # 把同步任务丢到线程池里执行，别阻塞事件循环
    await loop.run_in_executor(None, trigger_future)
    
    # 等待Future返回结果
    result = await my_future
    print(f"拿到Future结果啦: {result}")

3. 别让主事件循环被阻塞 #

如果你的监听脚本里用了time.sleep()这类同步阻塞函数，主事件循环会被卡死，initEvga里的Future任务根本没机会被调度执行。

正确做法：

• 所有需要等待的操作，都用asyncio.sleep()代替time.sleep()。
• 任何同步阻塞的任务，都用loop.run_in_executor()放到线程池里处理。

比如主函数应该这么写：

async def main():
    subscriber = pubsub_v1.SubscriberClient()
    subscription_path = subscriber.subscription_path(config.PROJECT_ID, config.SUBSCRIPTION_ID)
    
    async def callback(message):
        await initEvga(message, asyncio.get_running_loop())
        message.ack()
    
    streaming_pull_future = subscriber.subscribe(subscription_path, callback=callback)
    print(f"开始监听订阅: {subscription_path}")
    
    try:
        # 用asyncio.Future()让主循环一直跑，别用time.sleep()
        await asyncio.Future()
    except KeyboardInterrupt:
        # 收到中断信号时，优雅停止订阅
        streaming_pull_future.cancel()
        await streaming_pull_future

if __name__ == "__main__":
    asyncio.run(main())

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