Python 3.5中基于async/await实现异步数据流的非队列方案问询

当然可以！你完全可以借助Python的**异步生成器（Async Generator）**来实现这种逐流式的数据传递，不用依赖asyncio.Queue就能让C持续返回数据给B，B处理后再逐个传给A。这种方式更贴合异步协程的原生特性，代码更简洁，也避免了队列频繁读写的额外开销。

核心思路 #

异步生成器允许我们在async def函数里使用yield关键字，每次yield会返回一个值并暂停函数执行，当调用方通过async for迭代它时，函数会从暂停处继续执行，直到下一次yield或者函数结束。这样就能实现“生成一个、传递一个、处理一个”的流式流程。

修改后的完整代码 #

import asyncio

async def A():
    # 通过async for逐个接收B处理后的结果
    async for val in B():
        print(f"A 收到处理后的数据: {val}")
        # 这里可以添加A自己的业务逻辑

async def B():
    # 异步迭代C生成的原始数据
    async for raw_data in C():
        # 调用D处理数据，然后把处理结果yield给A
        processed_data = await D(raw_data)
        yield processed_data

async def C():
    i = 0
    while i < 5:
        await asyncio.sleep(1)
        yield i  # 用yield逐个返回数据，替代原来的return
        i += 1  # 别忘了递增计数器，避免无限循环

async def D(val):
    # 这里可以添加你的数据处理逻辑，示例直接返回原数加10模拟处理
    return val + 10

async def main():
    await A()

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

代码说明 #

1. C函数：从普通协程改成异步生成器，用yield i替代return i，每次循环都会返回当前的i，然后暂停等待下一次迭代。
2. B函数：同样改成异步生成器，通过async for迭代C的输出，每拿到一个原始数据就调用D处理，再把处理后的结果yield给A。
3. A函数：用async for迭代B的输出，逐个接收处理完成的数据，执行自己的业务逻辑。

为什么这比队列更合适？ #

• 原生支持，代码更简洁：不需要额外创建和维护队列对象，直接利用异步生成器的迭代特性，代码逻辑更直观。
• 无额外开销：避免了队列的put()/get()操作，减少了不必要的上下文切换和内存开销。
• 天然流式控制：数据是按需生成和传递的，不会提前缓存所有数据，更适合处理大量或持续生成的数据。

注意事项 #

• 必须用async for来迭代异步生成器，普通的for循环无法工作。
• 如果需要中途终止流式传递，可以在async for循环里使用break，Python会自动在生成器函数结束时抛出StopAsyncIteration异常来终止迭代。
• 如果生成器需要处理外部终止信号（比如用户中断），可以结合asyncio.Event或者try/finally块来实现优雅退出。

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