给定一个常规生成器,您可以从中获取一个迭代器,该迭代器只能使用一次并从上次中断的地方继续。像这样 -
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
最佳答案
常规生成器和异步生成器之间的唯一区别在于,异步生成器的 __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())
关于python - 从 Python 3 中的异步生成器继续,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/48934171/