从 Python 中的异步生成器继续 3
Continue from a Async Generator in Python 3
给定一个常规生成器,您可以从中获取一个迭代器,该迭代器只能使用一次并从您停止的地方继续。像这样-
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())
给定一个常规生成器,您可以从中获取一个迭代器,该迭代器只能使用一次并从您停止的地方继续。像这样-
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())