铁律 1：网络 I/O 必须 await#

（如：请求 OpenAI、查数据库、Redis、读写 S3）

• 为什么要 await？ 因为这些操作不费 CPU。CPU 只是发了个指令，然后就在那傻等。 如果你不用 await，CPU 就会像个傻子一样盯着网卡，什么都不干。用了 await，CPU 就能在等待期间去处理别人的请求。
• 做法：使用支持异步的库（httpx, asyncpg, motor），直接 await。

铁律 2：纯 CPU/内存计算，async 毫无用处#

（如：Pandas 处理 Excel、视频转码、哈希计算、超长 for 循环）

• 为什么 async 没用？ 因为 async 的本质是单线程的。 你把一个 for i in range(1亿) 放到 async def 里，它依然是在主线程跑。CPU 必须全神贯注地算这个加法，根本腾不出手去处理别人的请求（Event Loop 被卡死）。 在这种情况下，async 甚至比同步还慢一点点（因为有协程切换的开销，虽然微乎其微）。
• 做法：如果你直接写在 async def 里，服务器直接死锁。

铁律 3：解决 CPU 阻塞的唯一解药 —— 扔到“池子”里#

既然 CPU 必须干这个活，主线程又不能干，那就只能请外援。

方案 A：def + asyncio.to_thread (线程池) —— 最推荐 🌟#

这是 FastAPI 开发者的标准动作。

• 原理：利用 run_in_threadpool 机制。虽然 Python 有 GIL（全局锁），限制了多线程无法利用多核 CPU 并行计算，但是：
  1. 解放了 Event Loop：主线程（接待员）终于空闲了，可以继续响应 /ping 接口。
  2. Pandas 的特权：像 Pandas/Numpy 这种底层是 C 写的库，经常会在做重型计算或 I/O 时主动释放 GIL。这时候线程池是真的能并行跑的！
• 适用场景：Excel 解析、图片处理、大部分 Pandas 操作。

方案 B：ProcessPoolExecutor (进程池) —— 核武器 ☢️#

如果你的计算是纯 Python 代码（比如写了 3层嵌套 for 循环算素数），连 GIL 都绕不过去。

• 原理：开一个全新的 Python 进程。
• 缺点：开销大，数据传输（序列化）慢，不能共享内存。
• 适用场景：极度复杂的数学计算、加密解密、必须占满多核 CPU 的任务。

解决各大厂商同步的 sdk#

用 to_thread 包裹后的性能损耗（线程切换）相对于网络传输（几十毫秒）来说，是可以忽略不计的。

import asyncio
from functools import partial

class AsyncWrapper:
    def __init__(self, sync_obj):
        self.sync_obj = sync_obj

    def __getattr__(self, name):
        # 获取原对象的同步方法
        func = getattr(self.sync_obj, name)

        # 返回一个异步函数
        async def wrapper(*args, **kwargs):
            # 利用 partial 固定参数，扔进线程池
            return await asyncio.to_thread(func, *args, **kwargs)
        return wrapper

# --- 使用起来就舒服多了 ---
# 假设 bucket 是阿里云原本的同步对象
async_bucket = AsyncWrapper(auth.Bucket(..., 'my-bucket'))

# 直接 await，不需要手动写 to_thread 了
await async_bucket.put_object('file.txt', 'content')