在升级 Python 版本后,该问题通常会被解决。但若是不想升级 Python 版本,可以根据以下方案解决:
将代码中存在阻塞的方法,使用线程或进程池进行异步调用,比如使用 run_in_executor()
方法。
示例:
import asyncio
import concurrent.futures
async def main():
with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor:
loop = asyncio.get_event_loop()
tasks = [
loop.run_in_executor(executor, blocking_function_1, args_1),
loop.run_in_executor(executor, blocking_function_2, args_2)
]
await asyncio.gather(*tasks)
def blocking_function_1(args_1):
# 阻塞的方法1
def blocking_function_2(args_2):
# 阻塞的方法2
此外,也可以使用 asyncio.run()
方法替代 run_until_complete()
方法,它是 Python 3.7 中引入的新方法。
示例:
import asyncio
async def foo():
# 异步的代码逻辑
async def main():
await asyncio.gather(foo())
asyncio.run(main())