AnyIO 内存流已经 aclose，receive 为什么还在等待

关闭接收流后，另一个任务里已经开始的 await receive.receive() 却没有返回，退出 TaskGroup 时也一直等着它。给同一个流再调用一次 aclose()，通常不能解决这个等待。

AnyIO 4.15.1 的内存对象流存在这条边界：关闭接收端会标记它关闭，并在最后一个接收端关闭时唤醒等待中的发送者，却不会唤醒这个接收端自己先前挂起的接收操作。它发生在 asyncio 和 Trio 后端，不能靠换后端就认定已消除。

新调用已经关闭，旧调用仍可能挂起

重新发起接收时，receive_nowait() 会检查关闭标志并抛出 ClosedResourceError。但已经等待的接收操作此前已完成这项检查，此时在等内部事件，接收端的 close() 没有给它发出唤醒通知。

因此“新调用立刻报错，原任务还没退出”并不矛盾。先记录接收开始、关闭发生和任务退出的时刻，确认关闭的是哪一个流对象、是否有克隆、任务正在哪个 await 上等待。

aclose() 释放一个流端的使用权，不能作为中断其他任务的通用取消接口。把它包进 BufferedByteReceiveStream，也不能凭包装层保证底层挂起的读取会被打断。

正常结束生产：关闭全部发送端

如果含义是“没有更多数据了，但已发送的数据仍要处理”，应让所有发送端和它们的克隆在生产结束后关闭。接收端可以继续消费缓冲区，随后得到 EndOfStream，或让 async for 正常结束。

一个发送端关闭，但另一个克隆仍活着时，通道仍可能继续有数据。检查生产者异常退出后是否漏关克隆，比重复关闭接收端更有用。不要为了让消费者退出而丢弃仍应处理的数据。

需要立即中断消费者：取消拥有它的任务

若需求是放弃等待，不再处理剩余项，应使用明确归属的取消作用域。下面只展示单个消费者的生命周期；取消后不会承诺排空通道：

import anyio


async def consume(stream, *, task_status=anyio.TASK_STATUS_IGNORED):
    async with stream:
        with anyio.CancelScope() as scope:
            task_status.started(scope)
            async for item in stream:
                print(item)


async def main():
    send, receive = anyio.create_memory_object_stream[int](0)
    async with anyio.create_task_group() as group:
        consumer_scope = await group.start(consume, receive)
        consumer_scope.cancel()
    await send.aclose()


anyio.run(main)

接收任务自己进入取消作用域，启动握手把该作用域交给管理者；管理者取消后，等待中的异步接收可以在取消点退出，再由上下文关闭流。不要从另一个任务进入或退出这个作用域，也不要只关闭对象却继续等待消费者自行结束。

取消整个 TaskGroup 也可以结束它拥有的消费者，但同时会影响组内其他任务；只需停止一个消费者时，不要无意扩大取消范围。业务已有取消清理逻辑时，要保留它，并让取消继续传播，不能用宽泛异常捕获把等待重新启动。

改动前记录克隆的创建与关闭位置，分别核对正常生产结束、消费者被取消、生产者失败和缓冲区尚有数据的情况。成功标准是既定的排空或放弃策略得到执行、任务确实退出、没有遗留等待者。恢复旧实现前先处理仍在运行的生产者和消费者，避免两套消费流程同时使用同一通道。