asyncio场景下如何实现带超时的异步回调等待功能

你现有代码的核心问题：当前写法先await expensivefunction(callback=callback)会阻塞等待它10秒执行完成，后续的asyncio.wait_for完全不会起到超时作用，因为此时回调已经被调用完毕了。
另外你原代码中对异步回调的判断逻辑存在问题：inspect.isawaitable(callback)判断的是回调函数本身是否为可等待对象，实际上异步函数调用后的返回值才是可等待对象，因此正确的写法是先执行回调拿到返回值，再判断返回值是否可等待，下面的示例代码中已经修正了这个问题。

方案1：直接给耗时函数加超时（适配你当前的代码逻辑） #

你给出的示例中，expensivefunction已经内置了对回调的处理，因此expensivefunction执行完成的时机和回调执行完成的时机完全一致，直接把它放到asyncio.wait_for里即可，这是最简单的实现：

import asyncio, inspect

async def expensivefunction(callback):
    # 模拟耗时操作
    await asyncio.sleep(10)
    result = 10
    # 调用回调并判断是否需要等待
    cb_ret = callback(result)
    if inspect.isawaitable(cb_ret): 
        await cb_ret

# 示例异步回调
async def callback(result): 
    print(result)

async def myfunc():
    try:
        # 直接等待耗时函数完成，设置5秒超时
        await asyncio.wait_for(expensivefunction(callback=callback), timeout=5)
        print("回调执行完成")
    except asyncio.TimeoutError:
        print("超时5秒未完成")
        # 按需抛出异常
        raise

asyncio.run(myfunc())

适用场景：回调在expensivefunction末尾调用，且异步回调会被expensivefunction内部等待的场景。

方案2：用Future解耦回调等待（通用场景） #

如果你的回调和expensivefunction的生命周期是解耦的（比如expensivefunction会立刻返回，回调由后续其他事件触发），可以用asyncio.Future作为回调触发的通知载体，实现任意位置等待回调+超时控制：

import asyncio, inspect

async def expensivefunction(callback):
    await asyncio.sleep(10)
    result = 10
    cb_ret = callback(result)
    if inspect.isawaitable(cb_ret): 
        await cb_ret

async def callback(result): 
    print(result)

def wrap_callback(original_cb, future: asyncio.Future):
    """包装原始回调，执行完成后通知Future"""
    async def wrapper(result):
        try:
            # 执行原始回调逻辑
            cb_ret = original_cb(result)
            if inspect.isawaitable(cb_ret):
                await cb_ret
            # 回调执行成功，设置Future结果
            future.set_result(result)
        except Exception as e:
            # 回调执行异常，传递给Future
            future.set_exception(e)
    return wrapper

async def myfunc():
    # 创建Future用于接收回调完成的通知
    cb_future = asyncio.Future()
    # 包装原始回调
    wrapped_cb = wrap_callback(callback, cb_future)
    # 将耗时函数放到后台运行，不阻塞主逻辑的超时等待
    bg_task = asyncio.create_task(expensivefunction(wrapped_cb))
    try:
        # 等待回调完成，超时5秒
        result = await asyncio.wait_for(cb_future, timeout=5)
        print(f"回调执行完成，结果为{result}")
    except asyncio.TimeoutError:
        print("超时5秒未触发回调")
        # 取消后台的耗时任务，避免孤儿运行
        bg_task.cancel()
        # 等待任务取消完成，避免控制台抛未处理取消异常的警告
        try:
            await bg_task
        except asyncio.CancelledError:
            print("后台耗时任务已取消")
        # 按需抛出超时异常
        raise

asyncio.run(myfunc())

原理说明 #

• asyncio.Future是异步场景下用来标记单次异步事件完成的标准组件，我们可以手动控制它的完成状态，正好匹配“回调被触发”这个一次性事件。
• 包装回调的逻辑兼容了同步/异步两种回调类型，同时会把回调执行过程中的异常正确传递到等待侧，避免程序卡死。
• 后台Task的方式可以让耗时逻辑和超时等待并行执行，超时后主动取消后台任务避免资源泄漏。

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