Python contextlib.aclosing 怎么收口异步资源：退出顺序、异常传播与测试验收

流式读取器已经把最后一条记录交给调用方，连接却迟迟没有释放，问题往往不在读取循环，而在异步迭代器退出时没有走完清理方法。Python 的 contextlib.aclosing 可以把对象的 aclose() 放进 async with 的退出阶段，让正常结束、提前 break 和异常退出都落到同一条清理路径。

只要异步资源提供了可等待的 aclose()，就可以用 aclosing(resource) 把清理责任绑定到当前异步上下文；真正的验收重点是退出后清理计数、异常原样传播，以及清理异常是否被测试捕获。

先看一个提前结束却没有清理的现场

下面的资源故意保留一个异步关闭方法。它不是网络客户端的完整实现，只用来观察生命周期：读取两条数据后提前退出，如果没有显式调用 aclose()，closed 仍然是 False。

import asyncio


class EventStream:
    def __init__(self):
        self.closed = False
        self.events = ["ready", "changed", "done"]

    def __aiter__(self):
        return self

    async def __anext__(self):
        if not self.events:
            raise StopAsyncIteration
        await asyncio.sleep(0)
        return self.events.pop(0)

    async def aclose(self):
        await asyncio.sleep(0)
        self.closed = True


async def consume_without_cleanup():
    stream = EventStream()
    async for event in stream:
        print(event)
        if event == "changed":
            break
    return stream.closed


print(asyncio.run(consume_without_cleanup()))  # False

这里别急着把 break 当成资源关闭信号。async for 只负责迭代协议，不会替一个普通对象猜测它该调用哪个异步清理方法。资源的所有权必须由外层上下文明确接住。

用 aclosing 把退出顺序固定下来

aclosing 返回一个异步上下文管理器。进入 async with 时拿到原对象，离开代码块时等待它的 aclose()。因此，消费逻辑只需要决定何时退出，不必在每个 break、return 分支里重复补清理。

from contextlib import aclosing


async def consume_with_cleanup():
    stream = EventStream()
    async with aclosing(stream) as current:
        async for event in current:
            print(event)
            if event == "changed":
                break
    return stream.closed


print(asyncio.run(consume_with_cleanup()))  # True

验证时至少看两个结果：上下文退出后 stream.closed 必须为 True，同时 changed 之后没有继续消费 done。前者证明清理走了，后者证明清理没有替代业务的停止条件。

三个方案怎么选：显式调用、aclosing 与异步生成器管理

工程里常见的选择有三种。显式调用 await resource.aclose() 最直观，但每个提前退出点都要保证执行，异常分支尤其容易漏掉；自定义异步上下文管理器适合同时管理连接、指标和事务，但实现成本更高；aclosing 适合已经有 aclose()、只缺统一退出边界的对象。

这个选择的边界很清楚：aclosing 不会替你添加 aclose()，也不会处理资源之间的依赖顺序。如果关闭一个资源前必须先提交另一个对象，就应该把顺序写进专门的上下文管理器，而不是把多个对象随意嵌套。

异常退出时，先验收清理再判断异常

async with 的主体抛出异常后仍会进入退出阶段，所以清理动作不能只在“正常读完”时测试。下面用一个标记异常验证两件事：aclose() 被调用，主体异常仍然能被外层捕获。

async def consume_with_error():
    stream = EventStream()
    try:
        async with aclosing(stream):
            raise RuntimeError("decode failed")
    except RuntimeError as exc:
        return stream.closed, str(exc)


print(asyncio.run(consume_with_error()))  # (True, 'decode failed')

实际测试里不要只断言异常文本。更稳的写法是给资源加一个 close_calls 计数，分别覆盖自然耗尽、提前 break 和主体异常；如果 aclose() 自己也会失败，再单独记录清理异常的优先级，避免把两个问题混成一个断言。

测试验收：三条路径足够暴露大多数漏清理

可以把资源替换成测试双，保留 aclose() 的可观察状态。测试名称直接写行为，回归时比测试一整段内部实现更容易定位。

async def collect_path(stop_mode):
    stream = EventStream()
    try:
        async with aclosing(stream):
            async for event in stream:
                if stop_mode == "break" and event == "changed":
                    break
                if stop_mode == "error" and event == "changed":
                    raise ValueError("bad event")
    except ValueError:
        pass
    return stream.closed


async def test_all_paths():
    assert await collect_path("natural") is True
    assert await collect_path("break") is True
    assert await collect_path("error") is True


asyncio.run(test_all_paths())

如果测试失败，先区分是资源没有进入上下文，还是主体里保存了另一个未关闭的对象。aclosing 只负责它包住的那个对象，不能替调用链里临时创建的子资源兜底。

别把 aclosing 当成所有异步资源的万能开关

• 对象只有同步的 close() 时，不要直接套 aclosing，应使用匹配同步协议的管理方式。
• 对象的 aclose() 需要特定事件循环或线程时，要在同一上下文中创建和关闭。
• 关闭动作可能耗时或可取消时，把取消策略写入测试，不能只看主体是否抛错。
• 如果资源来自第三方库，先确认它的 aclose() 是否幂等，再决定是否允许业务代码手动提前关闭。

相关问题

async for 结束后还需要 aclosing 吗？

如果迭代对象本身有独立的异步清理责任，仍建议显式管理。自然耗尽只说明迭代结束，不代表对象的连接、缓存或后台任务已经释放。

aclosing 能包住异步生成器吗？

可以，只要对象提供可等待的 aclose()。提前结束生成器时，使用它能把生成器清理纳入当前上下文的退出阶段。

清理失败应该覆盖主体异常吗？

不要凭经验决定。先记录两类异常，再按资源安全性和调用方契约制定规则；测试要明确最终抛出什么、原始异常是否保留。

总结

contextlib.aclosing 的价值不是少写一行 await，而是把异步资源的所有权和退出边界固定在同一个上下文里。选型时先确认对象确实有 aclose()，落地后用自然退出、提前退出、主体异常三条路径验收；一旦资源之间存在顺序依赖，就升级为专门的异步上下文管理器。