Microsoft Agent Framework Skills 执行 Scripts（实战指南）匦
本文能帮你解决什么？
1. 搞懂FastAPI异步（async/await）到底在什么场景下能真正提升性能。
2. 掌握在FastAPI中正确使用多线程处理CPU密集型任务的方法。
3. 避开常见的坑（比如阻塞操作、数据库连接池耗尽、GIL限制）。
4. 获得可直接复用的代码片段和配置建议。
?? 主要内容脉络
一、问题与背景：为什么你的async可能“假生效”？
二、核心原理：ASGI、async/await与多线程的关系
三、实战演示：I/O密集型 vs CPU密集型任务的正确处理姿势
四、注意事项与进阶思考：那些容易翻车的点
一、问题与背景：为什么你的async可能“假生效”？
很多人以为，只要给FastAPI的路由函数加上async def，就自动获得了高并发能力。其实不然。FastAPI基于ASGI（异步服务器网关接口），它确实允许异步处理请求。但异步不等于多线程，更不等于性能无限提升。
它的核心是“非阻塞”：当一个请求在等待I/O（比如查数据库、调外部API）时，事件循环（Event Loop）会去处理其他请求，而不是干等着。这意味着，如果你的async函数里干的是CPU密集型的活儿（比如复杂的计算、图像处理），那它依然会阻塞整个事件循环，其他请求照样排队。
官方文档虽然说了FastAPI支持异步，但没明确告诉你：异步的优势仅限于I/O密集型场景。 这是我用真金白银的线上故障换来的教训。
二、核心原理：ASGI、async/await与多线程的关系
好，咱们先来理清几个关键概念：
?? ASGI（Asynchronous Server Gateway Interface）：这是FastAPI的底层协议。你可以把它想象成一个高效的餐厅调度系统。服务员（事件循环）负责接待顾客（请求），如果某位顾客点菜后需要等厨房做菜（I/O等待），服务员不会傻等，而是先去接待其他顾客。厨房做好菜会通知服务员，服务员再回来上菜。这样，一个服务员就能同时照顾多桌客人。
?? async/await：这是Python的语法糖，用来定义协程（Coroutine）。async def声明一个函数是“可暂停的”，await表示“在这里可以暂停，去干别的”。
?? 多线程/多进程：当你的任务主要是CPU密集型（比如大量数学计算）时，异步帮不上忙。这时就需要请出多线程或多进程，把计算任务分摊到多个CPU核心上去。FastAPI本身不直接管理线程，但我们可以利用Python的concurrent.futures或asyncio.to_thread来实现。
简单总结：I/O密集型用async，CPU密集型用多线程/多进程，混合型任务两者结合。
三、实战演示：I/O密集型 vs CPU密集型任务的正确处理姿势
接下来重点来了，怎么在代码里落实？
场景1：纯I/O密集型（推荐使用async）
比如调用外部API、查询数据库。这是async的主场。
import asyncio
from fastapi import FastAPI
import httpx # 异步HTTP客户端
app = FastAPI()
@app.get("/fetch-data")
async def fetch_data():
# 模拟并发调用三个外部API
async with httpx.AsyncClient() as client:
tasks = [
client.get("https://api.example.com/data1"),
client.get("https://api.example.com/data2"),
client.get("https://api.example.com/data3")
]
responses = await asyncio.gather(*tasks)
return {"results": [r.json() for r in responses]}
?? 这里千万别用同步的requests库，否则会阻塞事件循环。务必使用httpx或aiohttp这种异步客户端。
场景2：CPU密集型（必须用多线程/多进程）
比如图像处理、数据分析。这时候就得请出进程池。
from fastapi import FastAPI
from concurrent.futures import ProcessPoolExecutor
import time
import os
app = FastAPI()
# 创建进程池，工作进程数建议设为CPU核心数
executor = ProcessPoolExecutor(max_workers=min(32, os.cpu_count() or 1))
def cpu_intensive_task(n: int):
"""模拟CPU密集型任务，比如图像处理"""
time.sleep(n) # 这里用sleep模拟计算耗时
return f"Task {n} completed"
@app.get("/process-image")
async def process_image():
# 将阻塞函数提交到进程池，避免阻塞事件循环
future = executor.submit(cpu_intensive_task, 2)
result = future.result()
return {"result": result}
# 应用关闭时正确关闭进程池
@app.on_event("shutdown")
def shutdown_event():
executor.shutdown(wait=True)
?? 这里有个坑：线程池大小max_workers不是越大越好。设置太大反而会增加上下文切换开销。一般建议设置为CPU核心数+1。
场景3：混合型（async + 多线程）
实际项目中，很多任务既涉及I/O又涉及计算。这时可以结合两者。
import asyncio
from fastapi import FastAPI
from concurrent.futures import ThreadPoolExecutor
import httpx
app = FastAPI()
executor = ThreadPoolExecutor(max_workers=4)
async def fetch_url(client: httpx.AsyncClient, url: str):
"""异步获取数据"""
response = await client.get(url)
return response.json()
def heavy_computation(data: dict):
"""模拟CPU密集型计算"""
time.sleep(1) # 模拟计算
return {"processed": data}
@app.get("/complex-task")
async def complex_task():
# 步骤1：并发I/O（异步）
async with httpx.AsyncClient() as client:
data = await fetch_url(client, "https://api.example.com/data")
# 步骤2：CPU计算（扔到线程池）
loop = asyncio.get_event_loop()
result = await loop.run_in_executor(executor, heavy_computation, data)
return result
如果你用的是Python 3.9+，还可以用asyncio.to_thread让代码更简洁。
四、注意事项与进阶思考：那些容易翻车的点
再说几个容易出问题的地方，都是血泪史：
?? 阻塞操作绝对不能放在async函数里：比如time.sleep()、同步的数据库驱动（如psycopg2）、同步的文件读写等。要用await asyncio.sleep()、异步驱动（如asyncpg）和aiofiles替代。
?? 数据库连接池配置：异步环境下，数据库连接池的大小需要重新评估。我遇到过因为连接池太小，高并发下所有请求都在等连接，导致服务雪崩的情况。建议根据实际压力测试调整。
?? GIL（全局解释器锁）限制：Python的GIL会让多线程在纯CPU任务上效率打折。如果计算极其密集，考虑用multiprocessing启动多进程，但要注意进程间通信的成本。
?? Uvicorn配置：生产环境运行FastAPI，通常用Uvicorn。建议设置--workers（进程数）为CPU核心数，--loop uvloop（使用更高效的事件循环）。例如：
uvicorn main:app --workers 4 --loop uvloop --host 0.0.0.0 --port 8000
?? 监控与日志：异步环境下，错误栈可能不那么直观。一定要打好日志，尤其是耗时操作。可以用asyncio.create_task时附加错误回调，避免任务静默失败。痪砂谋泵