如何用Python异步同时运行多个阻塞函数？

我是Python协程新手，使用Python 3.8.6，希望借助asyncio库并行处理阻塞函数。现有一个阻塞函数，需传入数组inputs中的不同参数并行执行：

示例耗时：

blocking_function(5) - takes 5 seconds
blocking_function(3) - takes 3 seconds
blocking_function(2) - takes 2 seconds

我尝试了以下代码，但函数仍串行执行：

inputs = [5,3,2]
async def main():
    tasks = [asyncio.create_task(blocking_function(input)) for input in inputs]
    result = await asyncio.gather(*tasks)
    print(result)

代码运行无报错，但总耗时8秒，如何实现并行执行使总耗时缩短至5秒？

你的代码串行执行的核心原因是：blocking_function是阻塞函数，不会让出asyncio的事件循环。asyncio的协程并行依赖任务主动释放CPU（比如await异步操作），而阻塞函数会一直占用事件循环，导致其他任务无法执行。

要实现真正的并行，需要把阻塞函数放到线程池或进程池中运行，让asyncio在等待阻塞任务时可以切换到其他任务。

方法1：使用loop.run_in_executor（适配Python 3.8） #

Python 3.9+的asyncio.to_thread本质是封装了run_in_executor，3.8可以直接用这个方法把阻塞函数放到线程中执行：

import asyncio
import time

# 模拟阻塞函数
def blocking_function(t):
    time.sleep(t)
    return f"Done in {t}s"

async def main():
    inputs = [5,3,2]
    loop = asyncio.get_event_loop()
    # 将每个阻塞任务提交到默认线程池
    tasks = [loop.run_in_executor(None, blocking_function, t) for t in inputs]
    results = await asyncio.gather(*tasks)
    print(results)

if __name__ == "__main__":
    start = time.time()
    asyncio.run(main())
    print(f"Total time: {time.time() - start:.2f}s")

这段代码会让每个blocking_function在独立线程中运行，总耗时约5秒，和最长任务的耗时一致。

方法2：显式指定线程池 #

如果需要控制线程数量，可以显式创建ThreadPoolExecutor：

import asyncio
import time
from concurrent.futures import ThreadPoolExecutor

def blocking_function(t):
    time.sleep(t)
    return f"Done in {t}s"

async def main():
    inputs = [5,3,2]
    # 创建线程池，指定最大线程数
    with ThreadPoolExecutor(max_workers=3) as executor:
        loop = asyncio.get_event_loop()
        tasks = [loop.run_in_executor(executor, blocking_function, t) for t in inputs]
        results = await asyncio.gather(*tasks)
        print(results)

if __name__ == "__main__":
    start = time.time()
    asyncio.run(main())
    print(f"Total time: {time.time() - start:.2f}s")

额外说明 #

• asyncio本身是单线程异步模型，无法直接并行执行阻塞代码，必须借助线程/进程池将阻塞任务移出事件循环线程。
• 线程池适合IO密集型阻塞任务（如网络请求、文件读写）；如果blocking_function是CPU密集型任务，建议改用ProcessPoolExecutor（进程池），避免GIL限制。

内容的提问来源于stack exchange，提问作者Shreyas Kulkarni