asyncio.to_thread 是首选封装方式，它将同步调用移至后台线程执行，避免阻塞事件循环；不可仅加 async/await，因底层系统调用仍会同步阻塞。

python中如何把老旧第三方同步sdk封装为异步非阻塞库？

直接用 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，也不能持有对事件循环线程独占资源（如未加锁的全局计数器）

示例：

import asyncio
from legacy_sdk import SomeSyncClient
<p>client = SomeSyncClient(api_key="xxx")</p><p>async def async_upload(self, file_path: str):</p><div class="aritcle_card flexRow artxards">
											<div class="artcardd flexRow">
												<a class="aritcle_card_img" rel="nofollow" href="/xiazai/skill13014" title="Python 环境工具专业版"><img
														src="https://img.php.cn/upload/skill/001/246/273/6adaf20d3191a227.jpg" alt="Python 环境工具专业版" onerror="this.onerror='';this.src='/static/lhimages/moren/morentu.png'" ></a>
												<div class="aritcle_card_info flexColumn">
													<a rel="nofollow" href="/xiazai/skill13014" title="Python 环境工具专业版" class="overflowclass">Python 环境工具专业版</a>
													<p class="overflowclass">Python 环境工具专业版功能概述Python 环境工具专业版是一项面向实际任务的技能。它将相关步骤、工具调用和结果整理方式集中到统一流程中，帮助使用者更快完成目标并减少重复操作。核心要点使用时应结合输入条件选择合适的执行方式，核对必要参数、依赖环境与输出内容，并按原始要求处理异常情况。该技能适合需要稳定复用相关能力的场景，可作为自动化工作流的一部分，也便于后续检查、调整和扩展。执行完成后建议检查结果是否符合预期，并保留必要记录。使用与执行同时注意输入完整性、执行条件和输出可用性。从功能定位来看，该技能</p>
												</div>
												<a rel="nofollow" href="/xiazai/skill13014" title="Python 环境工具专业版" class="aritcle_card_btn flexRow flexcenter"><b></b><span>下载</span>
												</a>
											</div>
										</div><h1>✅ 正确：委托给后台线程</h1><pre class="brush:php;toolbar:false;">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) 包裹整个调用，再在外层 catch asyncio.TimeoutError 并决定是否重试
鉴权逻辑不要藏在 SDK 内部：提取出 _gen_auth_header() 等函数，确保其本身是纯函数（无副作用、无状态），方便在异步上下文中安全复用
避免在 to_thread 中做重试：否则一次超时可能拖住整个线程数秒，浪费线程池资源

真正容易被忽略的是：很多同步 SDK 的“连接池”是线程局部的（如 requests.Session），在 to_thread 中反复创建新 session 会导致连接复用失效、TIME_WAIT 暴增。应在主线程预热并复用 session 实例，再传入线程执行。