Python asyncio.to_thread 不是并发加速：线程池边界、取消语义与阻塞函数验证

接口已经改成 async def，请求延迟却还是会在某个同步函数上突然拉长，这种情况在接入旧文件库、压缩库或同步 SDK 时很常见。asyncio.to_thread() 可以把这类调用移到线程池，让事件循环继续接收任务，但它解决的是“别堵住当前线程”，不是“所有代码都并行变快”。

先用一个阻塞函数看清线程边界

准备一个故意等待的同步函数。它不需要依赖网络，也不会被平台环境干扰，最适合用来观察线程名、耗时和取消行为。

import asyncio
import threading
import time

def read_legacy_file(delay: float) -> str:
    time.sleep(delay)
    return f"{threading.current_thread().name}: ready"

async def main() -> None:
    started = time.perf_counter()
    result = await asyncio.to_thread(read_legacy_file, 0.2)
    elapsed = time.perf_counter() - started
    print(result, f"{elapsed:.3f}s")

asyncio.run(main())

运行时通常会看到类似 asyncio_0: ready 0.201s 的结果。main() 仍在事件循环线程中，真正睡眠的是另一个工作线程。这里的收益是事件循环没有被 time.sleep() 卡住，而不是等待时间凭空消失。

默认线程池不是无限的：先做一次排队实验

把 20 个任务同时交给 to_thread()，不要只看最终总耗时，还要记录每个任务什么时候真正开始。开始时间被分成几批，说明工作线程数量或下游资源已经成了瓶颈。

async def one_job(index: int) -> tuple[int, float]:
    started = time.perf_counter()
    await asyncio.to_thread(read_legacy_file, 0.15)
    return index, time.perf_counter() - started

async def batch() -> None:
    rows = await asyncio.gather(*(one_job(i) for i in range(20)))
    print("max latency:", max(cost for _, cost in rows))

asyncio.run(batch())

这段代码的关键不是某一台机器的固定线程数，而是“任务数、线程池容量、下游连接数”三者要一起匹配。同步数据库驱动、图片解码和文件扫描都可能在池外还有自己的限制。生产环境更稳妥的做法是先用信号量把并发窗口收窄：

limit = asyncio.Semaphore(8)

async def bounded_job(index: int) -> tuple[int, float]:
    async with limit:
        return await one_job(index)

8 不是通用标准答案，只是一个可验证调整的起点。把它当成可配置的参数项，用 p95 延迟、线程等待时间和下游错误率共同调整到合理区间。

取消协程不等于杀掉已经运行的线程

这是最容易被误读的特性。下面的实验会在 50 毫秒后取消等待方：

def long_sync_work() -> str:
    time.sleep(1.0)
    print("sync work finished")
    return "done"

async def cancel_demo() -> None:
    task = asyncio.create_task(asyncio.to_thread(long_sync_work))
    await asyncio.sleep(0.05)
    task.cancel()
    try:
        await task
    except asyncio.CancelledError:
        print("等待方已取消")
    await asyncio.sleep(1.1)

asyncio.run(cancel_demo())

调用方会很快得到 CancelledError，但稍后仍可能看到 sync work finished。线程中的同步函数没有通用且安全的强杀接口，因此函数内部要有自己的超时、分段检查或幂等设计。对不能被中断的写盘、提交和外部副作用，别把“取消协程”当成回滚保证。

异常、上下文和退出阶段要分别验证

to_thread() 会把同步函数抛出的异常带回等待它的协程，这一点可以直接用测试用例锁定；但服务关闭时，如果还有线程任务未收尾，进程退出时间仍可能被拖长。

import contextvars

request_id = contextvars.ContextVar("request_id", default="-")

def inspect_context() -> str:
    return request_id.get()

async def check_context() -> None:
    request_id.set("req-42")
    value = await asyncio.to_thread(inspect_context)
    assert value == "req-42"

asyncio.run(check_context())

建议给每个调用增加明确的业务超时，并在应用退出前停止接收新任务、等待允许完成的任务，再记录仍未结束的数量。不要在退出钩子里无限等待一个已经失联的同步 SDK。

上线前的最小检查清单

• 确认被移出的函数确实是阻塞 I/O，而不是大段 CPU 计算。
• 为批量调用设置并发上限，测出 p95 延迟而不是只看平均值。
• 单测异常传递、超时和取消；明确取消后外部副作用是否仍会发生。
• 检查线程函数是否携带请求上下文，以及退出阶段是否能正常收尾。

常见问题

asyncio.to_thread 能绕过 Python 的 GIL 吗？

不要把它当成 CPU 并行方案。它更适合释放事件循环，让同步 I/O 在其他线程等待；纯 Python 计算通常应评估进程、原生扩展或独立任务队列。

取消 to_thread 后，文件写入会立刻停止吗？

不会自动保证。取消的是等待方，已经开始的同步函数可能继续完成，所以写入逻辑需要幂等、临时文件和提交阶段设计。

批量调用 to_thread 为什么还是慢？

常见原因是线程池排队、下游连接池更小，或同步函数本身耗时较长。先记录开始时间和完成时间，再分别调整并发窗口与下游配置。

什么时候应该换成进程或任务队列？

当任务是长时间 CPU 计算、需要独立重试，或不能让请求生命周期承载它时，进程池或持久化任务队列更合适。

把 to_thread() 当成事件循环和旧同步代码之间的隔离层，边界就清楚了：它能搬走等待，不能消除资源限制；能取消等待，不能凭空终止副作用。上线前把这两条写进测试和监控，通常比继续加线程更有效。