Python asyncio.to_thread 实战:在协程里调用阻塞库不卡住事件循环写了个 FastAPI 接口,里面调了一个只有同步版本的 SDK(比如某个老牌的图像处理库或数据库驱动),压测时 QPS 惨不忍睹,而且并发一上来所有请求都卡住。问题不在那个库慢,而在于你在协程里直接调了阻塞函数,把整个事件循环给堵死了。这篇讲清楚为什么会堵,以及asyncio.to_thread怎么救。先搞懂:一个阻塞调用堵死所有协程asyncio 是单线程事件循环。协程之间靠await主动让出控制权来实现并发。一旦某个协程执行了一个不会 await 的阻塞调用(time.sleep、requests.get、同步文件读写、CPU 密集计算),事件循环就被这一个协程霸占,其他所有协程全都得排队等它。看个能复现的例子:importasyncioimporttimedefblocking_io():time.sleep(2)# 模拟阻塞的同步库调用returnresultasyncdefworker(name):print(f{name}start)resultblocking_io()# 直接调阻塞函数——灾难print(f{name}done:{result})asyncdefmain():starttime.perf_counter()awaitasyncio.gather(worker(A),worker(B),worker(C))print(ftotal:{time.perf_counter()-start:.1f}s)asyncio.run(main())你以为三个 worker 并发跑,2 秒就完事?实际输出总耗时6 秒——因为blocking_io根本没让出事件循环,三个协程串行执行。更糟的是,如果这是个 Web 服务,这 6 秒里连健康检查接口都无法响应,整个服务像是死了。asyncio.to_thread:把阻塞调用甩到线程池Python 3.9 加入的asyncio.to_thread就是干这个的:它把同步函数丢到一个后台线程执行,并返回一个可以await的协程。当前协程在等待期间会让出事件循环,其他协程照常跑。importasyncioimporttimedefblocking_io():time.sleep(2)returnresultasyncdefworker(name):print(f{name}start)# to_thread 把 blocking_io 扔到线程池执行,await 期间事件循环自由resultawaitasyncio.to_thread(blocking_io)print(f{name}done:{result})asyncdefmain():starttime.perf_counter()awaitasyncio.gather(worker(A),worker(B),worker(C))print(ftotal:{time.perf_counter()-start:.1f}s)asyncio.run(main())这次总耗时2 秒,三个阻塞调用真正并行了。原理很简单:GIL 在遇到time.sleep这类 I/O 阻塞时会释放,所以多个线程能真正同时等待。带参数怎么传?位置参数直接跟,关键字参数用 functools.partialasyncio.to_thread(func, *args, **kwargs)支持透传参数:importasyncioimportfunctoolsdeffetch(url,timeout5,retries3):...# 某个只有同步版本的 http 库returnf{url}okasyncdefmain():# 位置参数和关键字参数都能直接传r1awaitasyncio.to_thread(fetch,http://a.com,timeout10)# 如果你要预先绑定一批参数再传,用 functools.partial 更清晰callfunctools.partial(fetch,timeout10,retries5)r2awaitasyncio.to_thread(call,http://b.com)print(r1,r2)asyncio.run(main())关键坑一:to_thread 救不了 CPU 密集任务to_thread用的是线程,而 Python 有 GIL。对于I/O 阻塞(sleep、网络、磁盘),等待时 GIL 会释放,多线程有效;但对于纯 CPU 计算(大循环、图像编解码、加解密),GIL 不会释放,多个线程还是抢同一个核,不会加速,甚至因线程切换更慢。defcpu_heavy(n):returnsum(i*iforiinrange(n))# 纯计算,GIL 全程持有# 这样并发跑 4 个,不会快 4 倍,基本等于串行awaitasyncio.gather(*[asyncio.to_thread(cpu_heavy,10**7)for_inrange(4)])CPU 密集要用进程,应该走loop.run_in_executor配ProcessPoolExecutor:importasynciofromconcurrent.futuresimportProcessPoolExecutorasyncdefmain():loopasyncio.get_running_loop()withProcessPoolExecutor()aspool:# 用多进程绕开 GIL,CPU 密集才能真正并行resultsawaitasyncio.gather(*[loop.run_in_executor(pool,cpu_heavy,10**7)for_inrange(4)])print(results)关键坑二:默认线程池只有很少的线程to_thread底层用的是事件循环的默认ThreadPoolExecutor,它的线程数默认是min(32, os.cpu_count() 4)。如果你同时甩几百个阻塞调用进去,超出线程数的会排队,并发度被悄悄限制。想要更大或独立的池,自己建一个,用run_in_executor指定:importasynciofromconcurrent.futuresimportThreadPoolExecutorasyncdefmain():loopasyncio.get_running_loop()# 为一批阻塞 I/O 单独开一个 50 线程的池,不和默认池抢资源withThreadPoolExecutor(max_workers50)aspool:tasks[loop.run_in_executor(pool,blocking_io)for_inrange(50)]awaitasyncio.gather(*tasks)关键坑三:取消不会真正停掉线程里的函数await asyncio.to_thread(...)被取消时,协程会抛CancelledError,但已经在线程里跑着的那个同步函数不会被中断——它会一直跑到自己结束,只是结果被丢弃。所以to_thread里的函数最好本身支持超时(比如给 http 库设timeout),别指望外层asyncio.wait_for能把它掐断。小结asyncio 是单线程,协程里直接调阻塞函数会堵死整个事件循环,并发退化成串行。asyncio.to_thread(func, *args, **kwargs)(3.9)把阻塞调用甩到线程池,当前协程await时让出循环,I/O 阻塞型任务能真正并行。CPU 密集别用它——GIL 拦着,要用run_in_executorProcessPoolExecutor走多进程。默认线程池线程数有限(约 32),大批量并发要自建ThreadPoolExecutor。取消协程停不掉已在跑的线程函数,超时得靠函数自己的 timeout 参数。一句话记忆点:I/O 阻塞用 to_thread,CPU 密集用 ProcessPool,别把线程当成绕开 GIL 的银弹。