asyncio.to_thread 是首选封装方式，它将同步调用移至后台线程执行，避免阻塞事件循环；不可仅加 async/await，因底层系统调用仍会同步阻塞。
直接用 asyncio.to_thread 封装最稳妥，别试图给同步方法加 async def 声明——那只是“假装异步”，底层 socket.recv() 或 requests.get() 依然会卡死整个事件循环。
为什么不能只加 async/await 关键字？
同步 SDK 的阻塞本质不在语法，而在它调用的系统调用（如 socket.connect()、os.read()、time.sleep()）或第三方库（如 requests.Session().get()）。加了 async def 只是让函数返回协程对象，执行时仍会同步阻塞事件循环。
- 现象：FastAPI 接口在调用封装后的
async def legacy_sdk_call()时，其他并发请求全部卡住，响应时间飙升 - 根本原因：该方法内部仍调用了
urllib3.PoolManager.request()这类同步网络层 - 验证方式：在方法里插入
print(f"start: {asyncio.get_event_loop()}")和print("done")，会发现两个 print 之间无其他协程被调度
asyncio.to_thread 是首选封装方式
asyncio.to_thread 是 Python 3.9+ 官方提供的轻量线程池封装，它把同步调用移出事件循环线程，交由后台线程执行，主线程继续调度其他协程。相比手动管理 ThreadPoolExecutor，它更简洁、开销更低、自动处理异常传播。
- 适用场景：
json.loads()解析大响应体、legacy_sdk.upload_file()上传本地文件、psycopg2.connect()（旧版）、open(path).read() - 不适用场景：纯 CPU 密集型（如
sum(range(10**8))），应改用loop.run_in_executor(None, ...)配合ProcessPoolExecutor - 注意点：传入的函数不能依赖
threading.local，也不能持有对事件循环线程独占资源（如未加锁的全局计数器）
示例：
立即学习“Python免费学习笔记（深入）”；
import asyncio
from legacy_sdk import SomeSyncClient
<p>client = SomeSyncClient(api_key="xxx")</p><div class="aritcle_card flexRow">
<div class="artcardd flexRow">
<a class="aritcle_card_img" href="/xiazai/skill2806" title="Python Code Tester"><img
src="https://img.php.cn/upload/skill/000/000/081/178937292776471.jpg" alt="Python Code Tester" onerror="this.onerror='';this.src='/static/lhimages/moren/morentu.png'" ></a>
<div class="aritcle_card_info flexColumn">
<a href="/xiazai/skill2806" title="Python Code Tester">Python Code Tester</a>
<p>代码功能测试skill，根据用户需求搜索代码、生成测试用例、执行测试并修复问题</p>
</div>
<a href="/xiazai/skill2806" title="Python Code Tester" class="aritcle_card_btn flexRow flexcenter"><b></b><span>下载</span> </a>
</div>
</div><p>async def async_upload(self, file_path: str):</p><h1>✅ 正确：委托给后台线程</h1><pre class="brush:php;toolbar:false;"><code>return await asyncio.to_thread(client.upload, file_path)async def async_parse_response(self, raw_json: bytes):
✅ 正确：CPU-bound 但轻量，to_thread 足够
return await asyncio.to_thread(json.loads, raw_json)
状态共享与并发安全必须显式处理
老旧 SDK 往往依赖实例属性或模块级全局变量维护状态（如 token 缓存、连接重试计数、session ID）。在异步并发下，多个协程共用一个 client 实例时，这些状态极易被交叉覆盖。
- 典型错误：
self._last_token被两个并发请求同时更新，导致第二个请求携带过期 token 失败 - 推荐方案一（隔离）：用
contextvars.ContextVar存储请求级状态，例如request_id_var = ContextVar("request_id", default=None) - 推荐方案二（加锁）：对共享写操作加
asyncio.Lock，但仅限低频写、高频读场景；避免在to_thread内部加锁，因线程上下文不同 - 规避策略：初始化时为每个协程分配独立 client 实例（需确认 SDK 是否支持多实例，有些 SDK 内部硬编码单例）
重试、超时、鉴权等逻辑要统一收口
同步 SDK 通常自带简单重试（如 requests.adapters.Retry），但默认不兼容异步语义。若直接封装 client.call()，则重试过程会完全阻塞线程，且无法被协程取消。
- 正确做法：在
to_thread外层做控制，例如先用asyncio.wait_for(..., timeout=10)包裹整个调用，再在外层 catchasyncio.TimeoutError并决定是否重试 - 鉴权逻辑不要藏在 SDK 内部：提取出
_gen_auth_header()等函数，确保其本身是纯函数（无副作用、无状态），方便在异步上下文中安全复用 - 避免在
to_thread中做重试：否则一次超时可能拖住整个线程数秒，浪费线程池资源
真正容易被忽略的是：很多同步 SDK 的“连接池”是线程局部的（如 requests.Session），在 to_thread 中反复创建新 session 会导致连接复用失效、TIME_WAIT 暴增。应在主线程预热并复用 session 实例，再传入线程执行。
