asyncio as_yielded 来自异步生成器

我希望能够从许多异步协程中产生收益。异步的as_completed有点接近我正在寻找的东西（即我希望任何协程能够随时返回调用者然后继续），但这似乎只允许常规协程具有单个返回。

这是我到目前为止所拥有的：

import asyncio


async def test(id_):
    print(f'{id_} sleeping')
    await asyncio.sleep(id_)
    return id_


async def test_gen(id_):
    count = 0
    while True:
        print(f'{id_} sleeping')
        await asyncio.sleep(id_)
        yield id_
        count += 1
        if count > 5:
            return


async def main():
    runs = [test(i) for i in range(3)]

    for i in asyncio.as_completed(runs):
        i = await i
        print(f'{i} yielded')


if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    loop.run_until_complete(main())
    loop.close()

更换runs = [test(i) for i in range(3)] with runs = [test_gen(i) for i in range(3)]并为for i in asyncio.as_completed(runs)迭代每个产量就是我所追求的。

这是否可以用 Python 表达，是否有任何第三方可以为您提供比协程流程标准库更多的选择？

Thanks

您可以使用aiostream.stream.merge http://aiostream.readthedocs.io/en/latest/operators.html#aiostream.stream.merge:

from aiostream import stream

async def main():
    runs = [test_gen(i) for i in range(3)]
    async for x in stream.merge(*runs):
        print(f'{x} yielded')

在一个安全环境 http://aiostream.readthedocs.io/en/latest/core.html#aiostream.core.Stream.stream确保迭代后正确清理生成器：

async def main():
    runs = [test_gen(i) for i in range(3)]
    merged = stream.merge(*runs)
    async with merged.stream() as streamer:
        async for x in streamer:
            print(f'{x} yielded')

或者使用使其更紧凑pipes http://aiostream.readthedocs.io/en/latest/operators.html#pipe-lining:

from aiostream import stream, pipe

async def main():
    runs = [test_gen(i) for i in range(3)]
    await (stream.merge(*runs) | pipe.print('{} yielded'))

更多示例在文档 http://aiostream.readthedocs.io/en/latest/examples.html.

处理@nirvana-msu 评论

通过准备相应的源，可以识别产生给定值的生成器：

async def main():
    runs = [test_gen(i) for i in range(3)]
    sources = [stream.map(xs, lambda x: (i, x)) for i, xs in enumerate(runs)]
    async for i, x in stream.merge(*sources):
        print(f'ID {i}: {x}')