python - 从 Python 3 中的异步生成器继续

Question

给定一个常规生成器，您可以从中获取一个迭代器，该迭代器只能使用一次并从中断处继续。像这样 -

sync_gen = (i in range(10))
def fetch_batch_sync(num_tasks, job_list):
    for i, job in enumerate(job_list):
        yield job
        if i == num_tasks - 1:
            break
>>> sync_gen_iter = sync_gen.__iter__()
>>> for i in fetch_batch_sync(2, sync_gen_iter):
...     print i
... 
0
1
>>> for i in fetch_batch_sync(3, sync_gen_iter):
...     print i
... 
2
3
4

有没有办法用异步生成器做同样的事情？

async def fetch_batch_async(num_tasks, job_list_iter):
    async for i, job in enumerate(job_list_iter):
        yield job
        if i == num_tasks - 1:
            break

Answer 1

常规生成器和异步生成器之间的唯一区别是异步生成器的 __next__ 和 __iter__ 方法本身是异步的。这就是为什么普通的 for 和 enumerate 无法将它们识别为可迭代对象。

与常规生成器一样，可以从异步生成器中提取值的子集，但您需要使用适当的工具。 fetch_batch_async 已经使用了 async for，但它还应该使用 enemuerate 的异步版本；例如:

async def aenumerate(aiterable, start=0):
    i = start
    async for obj in aiterable:
        yield i, obj
        i += 1

fetch_batch_async 会像 enumerate 一样使用它:

async def fetch_batch_async(num_tasks, job_list_iter):
    async for i, job in aenumerate(job_list_iter):
        yield job
        if i == num_tasks - 1:
            break

最后，此代码使用 fetch_batch_async 从无限异步迭代器中提取多个项目:

import asyncio, time

async def infinite():
    while True:
        yield time.time()
        await asyncio.sleep(.1)

async def main():
    async for received in fetch_batch_async(10, infinite()):
        print(received)

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