Python asyncio 怎样把阻塞函数放到线程执行

在 asyncio 协程里调用同步文件读取、旧式 SDK 或阻塞式客户端时，最直接的做法是 await asyncio.to_thread(func, ...)。它把同步函数安排到线程执行，当前协程等待结果时，事件循环仍能处理其他任务。

但要先记住一个边界：取消等待 to_thread 的任务，只能让协程不再等待这个结果，不能凭空终止已经开始运行的同步函数。本文围绕这个边界比较 to_thread、run_in_executor 和协作式停止写法。

官方文档：https://docs.python.org/3.14/library/asyncio-task.html

先判断：真正阻塞的是哪一段

事件循环适合等待异步 I/O；如果协程直接调用同步函数，函数执行期间事件循环线程不能切换到其他任务。下面的 read_report 代表同步文件、旧客户端或阻塞式库调用，问题不在它“是不是写在 async 函数里”，而在函数本身没有把控制权交还给事件循环。

import asyncio
import time

def read_report(path: str) -> str:
    # 用阻塞睡眠模拟同步文件读取或旧式客户端调用。
    time.sleep(1)
    return f"loaded: {path}"

async def wrong() -> str:
    # 这个同步调用会占住事件循环线程，其他协程无法及时运行。
    return read_report("daily.csv")

async def main() -> None:
    # 正确判断前先区分：这里需要移出事件循环的是 read_report。
    result = await asyncio.to_thread(read_report, "daily.csv")
    print(result)

asyncio.run(main())

判断标准是“调用期间是否会同步等待”，不是函数名里有没有 async。如果库已经提供真正的异步客户端，应优先使用异步接口；to_thread 更适合暂时包住无法替换的同步 I/O。

最小写法：用 asyncio.to_thread 传参数并接住结果

asyncio.to_thread 接受位置参数和关键字参数，返回一个可以等待的协程。实际代码里可以直接 await，也可以交给 create_task 与其他异步任务并发等待。

import asyncio
from pathlib import Path

def read_text(path: str, *, encoding: str) -> str:
    # 同步文件 API 仍然阻塞，所以把它整体放到线程中。
    return Path(path).read_text(encoding=encoding)

async def load_two_files() -> tuple[str, str]:
    # 关键字参数由 to_thread 原样传给同步函数。
    first = asyncio.create_task(
        asyncio.to_thread(read_text, "config.toml", encoding="utf-8")
    )
    second = asyncio.create_task(
        asyncio.to_thread(read_text, "labels.txt", encoding="utf-8")
    )
    # 等待两个线程调用的结果，异常会回到这里。
    return await asyncio.gather(first, second)

asyncio.run(load_two_files())

这里的并发只表示两个同步调用不再轮流占住事件循环；线程池仍受可用工作线程和外部 I/O 限制。不要因为用了 to_thread 就无限创建任务，应按业务吞吐设置并发上限。

to_thread 和 run_in_executor 怎么选

两者都能把阻塞工作交给执行器，但抽象层级不同。to_thread 是面向常见 I/O 场景的高层入口，写法短，并会传播当前 contextvars.Context；run_in_executor 更底层，适合需要指定线程池、进程池或明确管理执行器的场景。

import asyncio
from concurrent.futures import ThreadPoolExecutor

def lookup(key: str) -> str:
    # 模拟一个只能同步调用的客户端查询。
    return f"value:{key}"

async def choose_executor() -> str:
    # 常规同步 I/O：高层 API 足够清晰。
    value = await asyncio.to_thread(lookup, "user:42")

    # 需要固定线程池时再使用底层接口，并显式关闭执行器。
    loop = asyncio.get_running_loop()
    with ThreadPoolExecutor(max_workers=4, thread_name_prefix="lookup") as pool:
        custom_value = await loop.run_in_executor(pool, lookup, "user:43")
    return f"{value}; {custom_value}"

asyncio.run(choose_executor())

如果只是“把一个同步 I/O 放到线程”，优先 to_thread；如果要复用自定义线程池、选择进程池，或代码已经围绕事件循环底层 API 组织，再用 run_in_executor。CPU 密集型 Python 代码通常不应仅凭换成线程就期待并行收益；要根据实现是否释放 GIL 和任务性质考虑进程池或其他方案。

contextvars、异常和资源清理要一起考虑

官方文档说明，to_thread 会把当前上下文变量传播到工作线程。因此日志请求号等上下文信息可以在同步函数里读取；线程函数抛出的异常则会在等待点重新抛出，调用方仍要按普通异常处理。

import asyncio
from contextvars import ContextVar

request_id = ContextVar("request_id", default="unknown")

def blocking_lookup() -> str:
    # to_thread 会传播当前 Context，线程里可读取请求号。
    if request_id.get() == "unknown":
        raise RuntimeError("missing request id")
    return f"lookup for {request_id.get()}"

async def handle() -> str:
    token = request_id.set("req-42")
    try:
        # 同步函数的异常会在 await 处回到协程。
        return await asyncio.to_thread(blocking_lookup)
    finally:
        # 恢复调用方上下文，避免请求号泄漏到后续逻辑。
        request_id.reset(token)

asyncio.run(handle())

线程函数内部创建的文件句柄、连接或临时资源仍应由函数自己用 with 或 try/finally 清理。把同步函数交给线程并不会改变资源所有权，也不会自动为第三方客户端增加线程安全能力。

取消任务时，为什么阻塞函数可能还在运行

取消的是外层 asyncio 任务。若线程里的同步函数已经开始执行，事件循环可以让等待协程收到 CancelledError，但不能安全地强制杀掉任意 Python 线程。于是会出现“调用方已经超时，后台函数仍可能继续”的结果。

import asyncio
import time

def slow_sync() -> str:
    # 这个函数没有停止检查，开始后不能被外层取消强行打断。
    time.sleep(3)
    return "finished"

async def bounded_call() -> None:
    task = asyncio.create_task(asyncio.to_thread(slow_sync))
    try:
        # 超时只约束协程等待的时间，不等于杀掉工作线程。
        async with asyncio.timeout(0.5):
            print(await task)
    except TimeoutError:
        print("caller timed out; sync work may still be running")

asyncio.run(bounded_call())

如果业务必须尽快停止，就让同步函数接收一个线程安全的停止信号，并在可接受的粒度上主动检查它。停止点应由被调用库支持；对一个不检查信号的黑盒函数，调用方只能停止等待并安排后续资源治理。

import asyncio
import threading
import time

def cancellable_job(stop: threading.Event) -> str:
    # 每轮检查停止信号，避免把取消误解为强制终止线程。
    for _ in range(10):
        if stop.is_set():
            return "stopped cooperatively"
        time.sleep(0.1)
    return "finished"

async def run_job() -> str:
    stop = threading.Event()
    task = asyncio.create_task(asyncio.to_thread(cancellable_job, stop))
    try:
        return await asyncio.wait_for(task, timeout=0.5)
    except TimeoutError:
        # 先发信号，再让线程函数自己走到安全停止点。
        stop.set()
        return await task

asyncio.run(run_job())

若超时后的返回值不重要，也可以在发出停止信号后记录任务状态并由后台治理；但必须确认函数会响应信号，不能把“取消 await”当作已经清理完成。

落地前按这张表做选择

最终可以把规则压缩成一句话：能用异步库就直接异步；不能替换的同步 I/O 用 to_thread；需要执行器控制才下沉到 run_in_executor；有停止要求就把停止信号设计进同步函数本身。

几个容易混淆的问题

to_thread 会自动让 CPU 密集任务变快吗？ 不会。CPython 下受 GIL 影响，普通 Python 字节码通常不因放到线程就获得真正的多核并行；扩展模块释放 GIL 或其他实现可能不同。

取消 await 后能立刻关闭线程吗？ 不能把它当成保证。外层任务可以停止等待，但已开始的同步函数需要自己结束或响应协作式停止信号。

为什么不用所有代码都 run_in_executor？ 它更底层，适合自定义执行器和进程池；常规同步 I/O 用 to_thread 能把意图写得更清楚。

to_thread 支持关键字参数吗？ 支持，函数后的 **kwargs 会直接传给同步函数。注意不要把协程对象误传给只接受同步调用的函数。