我正在尝试尽可能地写出习惯用语来从存储在字典中的未来结果中收集结果。
让我们假设我有以下代码:
import asyncio
async def sleep(seconds):
print(f'sleeping for {seconds} seconds')
await asyncio.sleep(seconds)
print(f'finished sleeping {seconds} seconds')
async def run():
tasks = {
'4': sleep(4),
'3': sleep(3),
'2': sleep(2),
'1': sleep(1),
}
print(await gather_from_dict(tasks))
if __name__ == '__main__':
asyncio.get_event_loop().run_until_complete(run())
我期望的输出是:
sleeping for 2 seconds
sleeping for 1 seconds
sleeping for 4 seconds
sleeping for 3 seconds
finished sleeping 1 seconds
finished sleeping 2 seconds
finished sleeping 3 seconds
finished sleeping 4 seconds
{'4': None, '3': None, '2': None, '1': None}
到目前为止,我发现最清晰的解决方案是:
async def gather_from_dict(tasks:Dict[Hashable, Awaitable],
loop=None, return_exceptions=False) -> Dict:
results = await asyncio.gather(
*tasks.values(),
loop=loop,
return_exceptions=return_exceptions
)
return dict(zip(tasks.keys(), results))
有没有更简单的方法实现这个?谢谢!!!
dict(zip(tasks.keys(), results))
,因为它有点耗时。 - Newskooler