如何用async def实现异步函数的细粒度资源管理？

解决方案代码 #

import asyncio
from typing import Callable, Awaitable, TypeVar

T = TypeVar('T')

async def resource_context(func: Callable[..., Awaitable[T]]) -> T:
    print("acquire")
    coro = func()
    await_iter = coro.__await__()
    try:
        while True:
            try:
                item = next(await_iter)
            except StopIteration as e:
                return e.value
            # 释放资源，准备等待
            print("release")
            # 等待当前awaitable完成
            result = await item
            # 重新获取资源，恢复执行
            print("acquire")
            # 将结果传递回协程，继续执行
            await_iter.send(result)
    finally:
        # 确保最终释放资源
        print("release")

测试示例 #

async def test_task():
    print("执行任务片段1")
    await asyncio.sleep(0.1)  # 此处会触发资源释放
    print("执行任务片段2")
    await asyncio.sleep(0.1)  # 再次触发资源释放
    return "任务完成"

async def main():
    result = await resource_context(test_task)
    print(result)

asyncio.run(main())

输出结果 #

acquire
执行任务片段1
release
acquire
执行任务片段2
release
acquire
release
任务完成

实现原理 #

这个实现完全复刻了原@asyncio.coroutine版本的细粒度控制逻辑：

1. 初始获取：先获取资源，再启动目标异步函数的协程迭代器
2. 每次await前释放：从协程迭代器拿到下一个要等待的对象时，先释放资源
3. 等待完成后重新获取：await对象完成后，立即重新获取资源，再把结果传递回协程，让目标函数继续执行
4. 最终释放：通过finally块确保无论协程正常结束还是抛出异常，资源都会被释放

和普通异步上下文管理器(async with)的区别：普通管理器只会在进入和退出时执行一次获取/释放，而此实现会在每次目标函数内部触发await时临时释放资源，函数执行代码逻辑时始终持有资源，完全满足细粒度资源管控需求。

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