Python异步生成器而不是异步生成器

问题描述

我的代码如下所示。我希望两个休眠可以共享相同的时间范围，并且需要1+2*3=7秒来运行脚本。 但是似乎出了点问题，所以仍然需要3*(1+2)秒。

是否知道如何修改代码？

import asyncio

async def g():
    for i in range(3):
        await asyncio.sleep(1)
        yield i

async def main():
    async for x in g():
        print(x)
        await asyncio.sleep(2)

loop = asyncio.get_event_loop()
res = loop.run_until_complete(main())
loop.close()

---

解决方案

async/await的要点是交错任务，而不是函数/生成器。例如，当您await asyncio.sleep(1)时，当前协程随睡眠一起延迟。同样，async for会将其协同例程延迟到下一项就绪。

为了运行单独的功能，您必须将每个部件创建为单独的任务。使用Queue在它们之间交换项目-只有在它们交换了项目之后，任务才会延迟。

from asyncio import Queue, sleep, run, gather


# the original async generator
async def g():
    for i in range(3):
        await sleep(1)
        yield i


async def producer(queue: Queue):
    async for i in g():
        print('send', i)
        await queue.put(i)  # resume once item is fetched
    await queue.put(None)


async def consumer(queue: Queue):
    x = await queue.get()  # resume once item is fetched
    while x is not None:
        print('got', x)
        await sleep(2)
        x = await queue.get()


async def main():
    queue = Queue()
    # tasks only share the queue
    await gather(
        producer(queue),
        consumer(queue),
    )


run(main())

---

如果您经常需要此功能，还可以将其放入包装异步迭代器的帮助器对象中。帮助器封装队列和单独的任务。您可以将帮助器直接应用于async for语句中的异步可迭代对象。

from asyncio import Queue, sleep, run, ensure_future


# helper to consume iterable as concurrent task
async def _enqueue_items(async_iterable, queue: Queue, sentinel):
    async for item in async_iterable:
        await queue.put(item)
    await queue.put(sentinel)


async def concurrent(async_iterable):
    """Concurrently fetch items from ``async_iterable``"""
    queue = Queue()
    sentinel = object()
    consumer = ensure_future(  # concurrently fetch items for the iterable
        _enqueue_items(async_iterable, queue, sentinel)
    )
    try:
        item = await queue.get()
        while item is not sentinel:
            yield item
            item = await queue.get()
    finally:
        consumer.cancel()


# the original generator
async def g():
    for i in range(3):
        await sleep(1)
        yield i


# the original main - modified with `concurrent`
async def main():
    async for x in concurrent(g()):
        print(x)
        await sleep(2)


run(main())