如何实现Python asyncio的「真正并发」协程？

问题根源 #

你之前的代码之所以是顺序执行效果，核心原因是协程内部没有真正的挂起点：如果await的是立即完成的操作（比如纯同步代码、无阻塞逻辑的async函数），事件循环没有机会切换到其他Task，只能等当前协程完全执行完毕后，再调度下一个。

解决方案 #

要实现你想要的并发效果，必须让await触发协程挂起。可以把同步阻塞的计算逻辑放到线程池执行，用asyncio.to_thread（Python3.9+）或者loop.run_in_executor包装，这样await该操作时，当前协程会释放事件循环，让其他Task有机会执行。

示例代码 #

错误的顺序执行代码

import asyncio

async def sum_task():
    print("start_sum")
    # 纯同步计算，无任何挂起点
    res = 0
    for i in range(10**7):
        res += i
    print("end_sum", res)

async def mul_task():
    print("start_mul")
    # 纯同步计算，无任何挂起点
    res = 1
    for i in range(1, 10**7):
        res *= i
    print("end_mul", res)

async def main():
    # 直接await协程，会强制顺序执行
    await sum_task()
    await mul_task()

asyncio.run(main())

修正后的并发代码

import asyncio

# 把同步阻塞的计算逻辑抽成普通函数
def sync_sum():
    res = 0
    for i in range(10**7):
        res += i
    return res

def sync_mul():
    res = 1
    for i in range(1, 10**7):
        res *= i
    return res

async def sum_task():
    print("start_sum")
    # await线程池中的任务，触发协程挂起
    res = await asyncio.to_thread(sync_sum)
    print("end_sum", res)

async def mul_task():
    print("start_mul")
    res = await asyncio.to_thread(sync_mul)
    print("end_mul", res)

async def main():
    # 先创建Task，让事件循环可以同时调度多个任务
    task1 = asyncio.create_task(sum_task())
    task2 = asyncio.create_task(mul_task())
    await task1
    await task2

asyncio.run(main())

关键要点 #

1. 必须创建Task：asyncio.create_task会把协程包装成可被事件循环调度的Task，只有这样才能实现多任务并发；直接await协程会强制顺序执行。
2. 提供有效挂起点：asyncio.to_thread将同步任务放到线程池执行，await该操作时，当前协程会释放事件循环，事件循环会立即切换到其他已创建的Task执行。
3. 低版本Python兼容方案：如果使用Python3.8及以下版本，可用loop.run_in_executor替代to_thread，效果完全一致：

async def sum_task():
    print("start_sum")
    loop = asyncio.get_running_loop()
    res = await loop.run_in_executor(None, sync_sum)
    print("end_sum", res)

内容的提问来源于stack exchange，提问作者Rubén