“纵有疾风来，人生不言弃”，这句话送给正在学习文章的朋友们，也希望在阅读本文《异步函数如何安全调用？anyio.to_thread使用指南》后，能够真的帮助到大家。我也会在后续的文章中，陆续更新文章相关的技术文章，有好的建议欢迎大家在评论留言，非常感谢！

async函数不能直接在同步代码中调用，需用anyio.to_thread.run_sync配合asyncio.run来桥接执行环境，即在新线程中启动新event loop运行async函数并同步返回结果。

async 函数不能直接在同步代码里调用，anyio.to_thread.run_sync 是正确入口

你写了个 async def fetch_data()，但当前上下文是纯同步的（比如 CLI 主函数、旧版 Flask 视图、或单元测试 setup），直接调用会报 RuntimeWarning: coroutine 'fetch_data' was never awaited，甚至返回一个悬空的 coroutine 对象。这不是语法错误，而是执行模型不匹配——你得把 async 工作“扔进线程”再等结果回来。anyio.to_thread.run_sync 就是干这个的：它在后台线程中启动 event loop，运行你的 async 函数，再把结果同步地交还给你。

注意：anyio.to_thread.run_sync 不是把 async 函数“变同步”，而是帮你桥接两种执行环境。它内部会创建新线程 + 新 asyncio event loop，所以开销比纯同步调用大，别在热路径频繁用。

必须用 await 包裹的 async 函数，不能直接传给 run_sync

run_sync 只接受同步可调用对象（即普通函数）。如果你把 async def 函数本身传进去，比如 run_sync(fetch_data)，会立刻失败——因为 fetch_data 返回的是 coroutine，不是可执行的同步函数。

正确做法是用 anyio.run_sync_in_worker_thread？不，那是旧名，已弃用。现在标准写法是：

• 用 lambda 或包装函数把 await 行为收束成同步调用：run_sync(lambda: anyio.run_sync_in_worker_thread(fetch_data))？不对，这嵌套错了。
• 正确方式是用 anyio.run_sync_in_worker_thread 的替代方案？等等，anyio 3.x 已统一为 to_thread.run_sync，但它只跑同步函数。所以你真正需要的是：anyio.run_sync_in_worker_thread 已被移除，现在必须用 anyio.from_thread.start_blocking_portal + portal.call？太重。
• 最简解法：用 asyncio.run 在同步函数里启动新 loop：def sync_wrapper(): return asyncio.run(fetch_data())，再传给 run_sync(sync_wrapper)。但注意：asyncio.run 禁止在已有 event loop 中调用（比如你已在 async def 里），而 to_thread.run_sync 启动的线程没有 loop，所以安全。

示例：

import anyio
import asyncio
async def fetch_data():
await asyncio.sleep(0.1)
return "done"
def sync_entry():
✅ 正确：在新线程里用 asyncio.run 执行 async 函数
result = anyio.to_thread.run_sync(
    lambda: asyncio.run(fetch_data())
)
return result

参数传递和异常传播要手动处理，run_sync 不自动解包 awaitable

run_sync 不会识别你传进去的函数返回的是 coroutine；它只管执行、拿返回值。所以如果你忘了用 asyncio.run 或等价机制，返回的就是一个未 await 的 coroutine 对象，后续调用 .result() 会报 AttributeError。

常见翻车点：

• 传了 fetch_data（函数对象）而不是 fetch_data()（调用结果）——后者是 coroutine，前者啥也不做。
• 用了 asyncio.get_event_loop().run_until_complete(fetch_data())，但在某些环境下（如 Windows + Python 3.12+）可能触发 RuntimeError: no running event loop，因为线程里 loop 没显式创建。
• async 函数抛异常时，会原样抛到 run_sync 调用处，无需额外 try，但堆栈会跨线程，可能丢失部分上下文。

建议封装一层防错：

def run_async_safely(coro_func, *args, **kwargs):
    async def _runner():
        return await coro_func(*args, **kwargs)
    return anyio.to_thread.run_sync(lambda: asyncio.run(_runner()))
用法
result = run_async_safely(fetch_data)

替代方案对比：为什么不用 trio.to_thread.run_sync 或 asyncio.to_thread？

anyio.to_thread.run_sync 是跨 runtime 的抽象，底层可选 asyncio 或 trio，但你得确保当前 anyio 后端一致。如果项目已固定用 asyncio，且你只是临时桥接，asyncio.to_thread（Python 3.9+）更轻量——但它只能在已有 asyncio event loop 中用，无法在纯同步主线程里调用（会报 RuntimeError: no current event loop）。

trio.to_thread.run_sync 同理，依赖 trio 运行时。

所以结论很实际：只要你在同步上下文（没 loop），又想用 anyio 生态，就老实用 anyio.to_thread.run_sync + asyncio.run 组合。别试图绕过“启动新 loop”这步——它不是冗余，是必要隔离。

最后提醒：如果 async 函数本身只做 CPU 密集型工作（比如解析大 JSON），别用这个模式，直接上 anyio.to_process.run_sync 或 concurrent.futures.ProcessPoolExecutor 更合适。IO 密集才值得走 event loop 路线。

本篇关于《异步函数安全调用方法及anyio.to_thread使用教程》的介绍就到此结束啦，但是学无止境，想要了解学习更多关于文章的相关知识，请关注golang学习网公众号！