Python异步编程实战：从event loop到高并发API优化

1. 项目概述：这不是一本“讲完就忘”的异步教程，而是一份我用三年踩坑、重写、再上线的实战手记

“Python Async Programming: The Complete Guide”——这个标题听起来像某本畅销书的副标题，但在我这儿，它是一份从生产环境里血淋淋抠出来的操作日志。不是理论推演，不是语法罗列，而是我在为一家日均处理23万次API调用的SaaS后台做性能优化时，亲手把requests全换成httpx、把Flask硬切到Starlette、把数据库连接池从psycopg2迁移到asyncpg后，每天盯着Grafana看P99延迟曲线从840ms砸到97ms的真实记录。核心关键词就三个：async/await、event loop、concurrency model——它们不是Python的“新功能”，而是你理解现代Web服务吞吐瓶颈的三把解剖刀。如果你正被“为什么加了多线程反而更慢”、“为什么Celery任务队列积压”、“为什么本地跑得飞快，一上K8s就超时”这类问题卡住，这篇内容就是为你写的。它不假设你懂协程调度器，也不要求你背过asyncio.run()的源码；它只假设你写过def get_user(id): return db.query(...)，并开始怀疑——这行代码，在并发1000时，到底在CPU上干了什么？又在IO等待时，让整个线程挂起了多久？适合谁？适合所有写过import asyncio却不敢在main()里加await的中级开发者；适合正在把Django REST Framework接口改造成异步的后端工程师；也适合想搞懂uvloop为什么比默认event loop快2.3倍的运维同学。它不教你怎么“学会异步”，而是带你亲手拆开Python解释器的IO等待机制，看清每一毫秒花在哪。

2. 整体设计思路：为什么必须放弃“多线程=并发”的思维定式？

2.1 传统阻塞模型的致命伤：一个请求卡死，整条流水线停摆

我们先看一个最典型的反面案例。这是我在2021年接手的老系统里真实存在的代码：

# sync_api.py —— 每个请求都同步调用外部天气API + 本地数据库查询 import requests import sqlite3 def get_weather_and_user(city: str, user_id: int): # 步骤1：同步HTTP请求（平均耗时1.2s，网络抖动时可达5s） weather = requests.get(f"https://api.weather.com/v3/weather/forecast?city={city}").json() # 步骤2：同步数据库查询（平均耗时15ms） conn = sqlite3.connect("users.db") cursor = conn.cursor() cursor.execute("SELECT * FROM users WHERE id = ?", (user_id,)) user = cursor.fetchone() conn.close() return {"weather": weather, "user": user}

这段代码在单用户测试时毫无问题。但当QPS升到120时，服务器负载直接飙到16+，top里看到的是大量python进程处于D（不可中断睡眠）状态——它们全卡在recvfrom()系统调用上，等天气API的TCP响应包。关键点来了：Python的CPython解释器有GIL（全局解释器锁），但GIL只锁住CPU计算，不锁IO等待。也就是说，当线程A在等网络响应时，GIL是释放的，理论上其他线程可以抢CPU。可现实是：你的Web服务器（比如uWSGI）默认只开4个工作进程，每个进程配8个线程，总共32个并发槽位。一旦这32个线程全被IO卡住，新来的请求只能排队等——这就是为什么你加了线程池，QPS却上不去。我实测过：当天气API响应时间从1.2s恶化到3.5s时，系统吞吐量断崖式下跌68%，错误率从0.02%跳到12.7%。这不是代码写得烂，而是模型错了：你用“抢占式多任务”的工具，去解决“协作式IO等待”的问题，天然不匹配。

2.2 异步模型的底层逻辑：用单线程+事件循环，榨干每一次IO空闲

Async编程的本质，是把“等待IO”这件事，从“线程挂起”变成“注册回调”。我们用asyncio重写上面的逻辑：

# async_api.py —— 同一事件循环内并发处理 import asyncio import httpx import asyncpg async def fetch_weather(client: httpx.AsyncClient, city: str) -> dict: # 不是阻塞等待，而是向event loop“预约”：等这个HTTP响应回来，就执行后续 response = await client.get(f"https://api.weather.com/v3/weather/forecast?city={city}") return response.json() async def fetch_user(pool: asyncpg.Pool, user_id: int) -> dict: # 同样，数据库查询也变成“预约” async with pool.acquire() as conn: return await conn.fetchrow("SELECT * FROM users WHERE id = $1", user_id) async def get_weather_and_user_async(city: str, user_id: int): # 创建一个HTTP客户端和数据库连接池（复用，非每次新建） async with httpx.AsyncClient() as client: pool = await asyncpg.create_pool("postgresql://...") # 并发发起两个IO操作：天气API + 数据库查询 # 注意：这里用asyncio.gather，不是await逐个执行！ weather_task = fetch_weather(client, city) user_task = fetch_user(pool, user_id) weather, user = await asyncio.gather(weather_task, user_task) await pool.close() return {"weather": weather, "user": user}

这段代码的核心转变在于：它不再需要32个线程来同时“等”32个IO响应，而只需要1个线程+1个事件循环，就能管理3200个并发IO任务。原理很简单：当await client.get(...)执行时，Python解释器不是让线程睡着，而是把当前协程（coroutine）的状态（比如局部变量、执行位置）保存下来，然后告诉event loop：“等这个socket有数据可读时，再唤醒我”。接着，event loop立刻去检查其他协程——比如另一个正在等数据库响应的协程，发现它的socket也ready了，就切换过去执行。这种“主动让出控制权”的模式，叫协作式多任务（cooperative multitasking）。它没有线程切换开销（无上下文切换、无锁竞争），内存占用极低（一个协程栈仅几KB，而线程栈默认1MB）。我部署后监控数据：同样QPS 120，CPU使用率从78%降到21%，内存占用从2.4GB压到680MB，P99延迟稳定在112ms±5ms。这不是魔法，是把硬件资源用到了刀刃上——CPU不浪费在“等”上，全用来“算”。

2.3 方案选型决策树：为什么选asyncio而非Trio/curio？为什么用httpx不用aiohttp？

在落地前，我和团队花了两周做技术选型验证。不是看文档吹得多好，而是拿真实业务场景压测。我们对比了三套方案：

结论很清晰：asyncio是唯一能零成本接入现有Django/Flask生态的标准方案；httpx比aiohttp快17%且API更接近requests，降低迁移成本；asyncpg是目前Python最快的PostgreSQL异步驱动，比aiomysql快3.2倍（我们测过10万行JOIN查询）。有人问为什么不选uvloop？答案是：我们选了，但它只是asyncio的event loop替换，不是独立框架。uvloop用Cython重写了select/epoll底层，实测在Linux上比默认SelectorEventLoop快2.3倍——这正是我们最终上线的配置。选型不是追求最新潮，而是找那个在“性能-学习成本-维护性-生态支持”四边形中，离你业务重心最近的那个顶点。

3. 核心细节解析：async/await不是语法糖，是运行时契约

3.1async def的三个隐藏契约：你写的不是函数，是协程工厂

很多开发者以为async def只是加了个async关键字的函数。错。它定义的是一个协程工厂（coroutine factory），调用它返回的不是结果，而是一个协程对象（coroutine object），这个对象必须被event loop驱动才能执行。看这个经典陷阱：

async def bad_example(): return "hello" # ❌ 错误：直接调用，得到的是coroutine对象，不是字符串！ result = bad_example() # <coroutine object bad_example at 0x...> print(result) # 输出：<coroutine object bad_example at 0x...>，不是"hello" # ✅ 正确：必须await，或用asyncio.run() result = await bad_example() # 在async函数内 # 或 result = asyncio.run(bad_example()) # 在同步代码中启动event loop

为什么这样设计？因为async def函数体内可能包含多个await，每次await都会让出控制权。Python需要在运行时保存/恢复协程的执行上下文（locals、指令指针等）。这个过程由C层的PyCoroObject结构体完成，开销远小于线程创建。我画了个简化的执行流图（文字版）：

调用 async def func() ↓ 返回 coroutine object（未执行！） ↓ await coroutine_object 或 asyncio.run(coroutine_object) ↓ event loop 将其加入就绪队列 ↓ 当轮到它执行时： - 恢复上次yield处的locals和指令指针 - 继续执行直到下一个await或return ↓ return值被传递给await表达式

这个机制决定了：所有await表达式右侧，必须是一个“可等待对象（awaitable）”——即实现了__await__方法的对象。async def函数返回的协程对象是awaitable；asyncio.Future是awaitable；httpx.AsyncClient.get()返回的也是awaitable。但普通requests.get()返回的Response对象不是awaitable，所以你不能await requests.get(...)——这是新手最常见的报错TypeError: object Response can't be used in 'await' expression的根源。

3.2 Event Loop：Python异步世界的“交通警察”，不是可有可无的背景板

asyncio的event loop不是装饰品，它是整个异步程序的中枢神经系统。它负责三件事：IO事件监听（epoll/kqueue）、定时器调度、协程调度。你写的每行await，最终都变成对event loop的一次“注册”或“唤醒”。关键认知：在同一个Python进程中，只能有一个正在运行的event loop（除非你手动创建多个线程，每个线程跑自己的loop，但这是高级玩法，99%的场景不需要）。

常见误区是认为asyncio.run()会“自动管理”一切。其实它做了三件关键事：

1. 创建一个新的asyncio.EventLoop实例（默认是ProactorEventLoopon Windows,SelectorEventLoopon Linux/macOS）
2. 把传入的协程对象注入loop，并运行它直到完成
3. 关闭loop并清理所有资源（包括取消未完成的task）

这意味着：你不能在asyncio.run()内部再调用asyncio.run()——会报RuntimeError: asyncio.run() cannot be called from a running event loop。这也是为什么FastAPI这类框架，自己管理loop生命周期，你只需写路由函数，不用操心run()。

我遇到过最痛的教训：在Celery任务里写异步代码。Celery worker是同步进程，它调用你的task函数时，event loop根本没启动。如果你在task里写await some_async_func()，会直接报错。解决方案只有两个：要么把task改成同步（用asyncio.run()包装），要么用Celery 5.0+的@app.task(autoretry_for=(Exception,), retry_kwargs={'max_retries': 3})配合asyncio.to_thread()把异步IO转成线程执行——但后者失去了异步的意义。所以我的经验是：异步代码必须生长在异步容器里（如FastAPI/Uvicorn、Starlette、Tornado async mode），不要试图把它塞进同步框架的缝隙中。

3.3 并发（Concurrency）与并行（Parallelism）：90%的性能问题源于混淆这两者

这是异步编程里最常被误解的概念。用一个生活化类比：

• 并行（Parallelism）= 你请了3个厨师，每人同时炒一道菜（多CPU核心同时执行不同任务）
• 并发（Concurrency）= 你一个人，左手煎蛋、右手煮面，听到水开就关火、看到蛋熟就翻面（单CPU核心快速切换任务）

asyncio解决的是并发IO密集型任务，不是并行CPU密集型任务。当你await asyncio.sleep(1)，你不是在让CPU空转1秒，而是在告诉event loop：“1秒后唤醒我，这期间你可以去干别的”。但如果你写await cpu_intensive_task()，而这个task内部全是for i in range(10**9): x += i*i，那它会霸占CPU整整几秒，event loop无法切换，所有其他协程都被饿死——这就是所谓的“协程饥饿（coroutine starvation）”。

解决方案就一个：把CPU密集型操作扔到线程池或进程池里执行，让它不阻塞event loop。asyncio提供了loop.run_in_executor()：

import asyncio import concurrent.futures def cpu_heavy_calculation(n: int) -> int: # 纯CPU计算，会阻塞 result = 0 for i in range(n): result += i * i return result async def safe_cpu_task(n: int): # 在默认线程池中执行，不阻塞event loop loop = asyncio.get_running_loop() with concurrent.futures.ThreadPoolExecutor() as pool: result = await loop.run_in_executor(pool, cpu_heavy_calculation, n) return result # 调用 result = await safe_cpu_task(10**7) # 安全！

我实测过：一个纯CPU任务在await中执行，会让1000个并发HTTP请求的P99延迟从110ms飙升到2.3s；而用run_in_executor后，延迟回归110ms±8ms。记住这个铁律：async/await的黄金领域是“等待”，不是“计算”。把计算交给线程/进程，把等待交给协程，各司其职，才是高性能的正道。

4. 实操过程：从零搭建一个高并发天气聚合API

4.1 环境准备与依赖锁定：为什么pyproject.toml比requirements.txt更适合异步项目

我们抛弃pip install -r requirements.txt这种粗放方式，改用pyproject.toml（PEP 621标准）。原因很实际：异步生态版本耦合极深。比如httpx>=0.23.0要求anyio>=3.6.0，而anyio<4.0.0又和uvloop>=0.17.0有兼容问题。手动锁版本容易漏掉间接依赖。pyproject.toml能声明精确版本和可选依赖：

# pyproject.toml [build-system] requires = ["hatchling"] build-backend = "hatchling.build" [project] name = "weather-aggregator" version = "0.1.0" dependencies = [ "httpx[http2]>=0.24.0", # 支持HTTP/2，提升长连接复用率 "asyncpg>=0.28.0", # PostgreSQL异步驱动 "fastapi>=0.104.0", # 异步Web框架 "uvicorn[standard]>=0.23.0", # ASGI服务器，带uvloop支持 "python-dotenv>=1.0.0", # 环境变量管理 ] [project.optional-dependencies] dev = [ "pytest-asyncio>=0.21.0", # 异步测试 "mypy>=1.5.0", # 类型检查 ]

安装时用pip install -e ".[dev]"，确保开发环境和生产环境依赖完全一致。特别注意httpx[http2]——我们实测开启HTTP/2后，与天气API的连接复用率从32%提升到89%，减少了TCP握手开销。而uvicorn[standard]会自动安装uvloop和httptools，比纯Python实现快3倍以上。部署时，我们用uvicorn main:app --host 0.0.0.0:8000 --workers 4 --loop uvloop --http httptools，其中--workers 4对应4个CPU核心，每个worker内跑一个uvloop，形成真正的多进程+单线程异步模型。

4.2 核心模块拆解：如何设计可测试、可监控、可降级的异步服务

一个健壮的异步服务不能只写async def handler()。我们按生产级标准拆成四层：

4.2.1 数据访问层（DAL）：用asyncpg.Pool管理连接，拒绝每次请求新建连接

# dal/weather.py import asyncpg from typing import Optional, Dict, Any class WeatherService: def __init__(self, pool: asyncpg.Pool): self.pool = pool async def get_forecast(self, city: str) -> Optional[Dict[str, Any]]: # 使用prepared statement防止SQL注入，且预编译提升性能 stmt = await self.pool.prepare( "SELECT data FROM weather_cache WHERE city = $1 AND updated_at > NOW() - INTERVAL '1 hour'" ) row = await stmt.fetchrow(city) if row: return row["data"] # 缓存未命中，调用外部API（见下节） return await self._fetch_from_api(city) async def _fetch_from_api(self, city: str) -> Dict[str, Any]: # 实际调用HTTP客户端，此处省略 pass

关键点：asyncpg.Pool是连接池，不是单个连接。它内部维护一个连接队列，pool.acquire()会从池中取一个空闲连接，用完自动归还。我们设min_size=10, max_size=100，避免连接数爆炸。实测表明，连接池大小设为2 * CPU核心数时，吞吐量最优——太少会排队，太多会触发数据库连接数限制。

4.2.2 HTTP客户端层：用httpx.AsyncClient的limits和timeout做熔断

# clients/weather_api.py import httpx from typing import Dict, Any class WeatherApiClient: def __init__(self): # 关键配置：限制并发连接数，防打爆上游 limits = httpx.Limits( max_connections=100, # 总连接数上限 max_keepalive_connections=20, # 长连接保活数 keepalive_expiry=60.0, # 长连接存活时间 ) # 全局超时：连接10s，读写30s，总耗时60s timeout = httpx.Timeout( connect=10.0, read=30.0, write=30.0, pool=60.0, ) self.client = httpx.AsyncClient( base_url="https://api.weather.com/v3/", limits=limits, timeout=timeout, http2=True, ) async def get_forecast(self, city: str) -> Dict[str, Any]: try: response = await self.client.get( "weather/forecast", params={"city": city}, # 请求级超时，覆盖全局 timeout=httpx.Timeout(20.0), ) response.raise_for_status() return response.json() except httpx.TimeoutException: # 降级：返回缓存或默认值 return {"error": "timeout", "fallback": "cached_data"} except httpx.HTTPStatusError as e: # 处理4xx/5xx raise WeatherAPIError(f"API error {e.response.status_code}")

这里limits和timeout是生命线。没有limits，1000并发请求会瞬间创建1000个TCP连接，上游API直接503；没有精细timeout，一个慢请求会拖垮整个event loop。我们线上配置后，API错误率从12.7%降到0.3%，其中92%的失败是可控的超时降级，而非雪崩。

4.2.3 业务逻辑层：用asyncio.gather和asyncio.wait控制并发粒度

# services/weather_service.py import asyncio from typing import List, Dict, Any from dal.weather import WeatherService from clients.weather_api import WeatherApiClient class AggregatedWeatherService: def __init__(self, weather_dal: WeatherService, api_client: WeatherApiClient): self.weather_dal = weather_dal self.api_client = api_client async def get_multi_city_forecast(self, cities: List[str]) -> List[Dict[str, Any]]: # 方案1：全部并发（适合城市数≤50） tasks = [self._get_single_city(c) for c in cities] return await asyncio.gather(*tasks, return_exceptions=True) async def _get_single_city(self, city: str) -> Dict[str, Any]: try: # 先查缓存 cached = await self.weather_dal.get_forecast(city) if cached: return {"city": city, "data": cached, "source": "cache"} # 缓存未命中，调用API api_data = await self.api_client.get_forecast(city) return {"city": city, "data": api_data, "source": "api"} except Exception as e: return {"city": city, "error": str(e), "source": "error"} async def get_multi_city_forecast_with_limit(self, cities: List[str], max_concurrent: int = 10) -> List[Dict[str, Any]]: # 方案2：分批并发（适合城市数>50，防压垮自身） results = [] for i in range(0, len(cities), max_concurrent): batch = cities[i:i+max_concurrent] batch_tasks = [self._get_single_city(c) for c in batch] batch_results = await asyncio.gather(*batch_tasks, return_exceptions=True) results.extend(batch_results) return results

asyncio.gather是并发利器，但要注意：它会同时启动所有task。如果cities列表有1000个城市，gather会瞬间发起1000个HTTP请求，即使你设了httpx.Limits，event loop也要处理1000个协程调度。所以我们在高负载场景用分批（batching）策略，用max_concurrent=10控制节奏。return_exceptions=True很重要——它让gather在某个task抛异常时不中断整个列表，而是把异常对象放进结果列表，方便后续统一处理。

4.2.4 API路由层：用FastAPI的BackgroundTasks做异步日志和监控

# main.py from fastapi import FastAPI, BackgroundTasks, HTTPException, Depends from services.weather_service import AggregatedWeatherService from typing import List, Dict, Any app = FastAPI(title="Weather Aggregator API") # 依赖注入：FastAPI自动管理生命周期 @app.on_event("startup") async def startup_event(): # 初始化连接池和客户端 app.state.weather_dal = WeatherService(await asyncpg.create_pool("...")) app.state.api_client = WeatherApiClient() @app.on_event("shutdown") async def shutdown_event(): await app.state.weather_dal.pool.close() await app.state.api_client.client.aclose() @app.get("/forecast/multi") async def multi_city_forecast( cities: List[str], background_tasks: BackgroundTasks, service: AggregatedWeatherService = Depends(get_weather_service), ) -> Dict[str, Any]: # 记录请求日志（异步，不阻塞主流程） background_tasks.add_task(log_request, cities) try: # 执行核心业务 results = await service.get_multi_city_forecast_with_limit(cities) return {"status": "success", "data": results} except Exception as e: # 统一错误处理 raise HTTPException(status_code=500, detail=f"Internal error: {str(e)}") async def log_request(cities: List[str]): # 异步写日志，用aiofiles或直接发到日志服务 import aiofiles async with aiofiles.open("/var/log/weather_api.log", mode="a") as f: await f.write(f"[{datetime.now()}] Requested {len(cities)} cities\n")

BackgroundTasks是FastAPI提供的轻量级异步任务队列，适合日志、指标上报、通知等“fire-and-forget”场景。它不保证执行成功（失败就丢弃），但绝不阻塞主请求流。我们用它把每条请求记录异步写入文件，避免磁盘IO拖慢API响应。

4.3 部署与监控：用Prometheus+Grafana盯死event loop的每一毫秒

异步服务的监控和同步服务完全不同。我们重点关注三个指标：

1. Event Loop Lag：event loop处理一个任务的平均延迟。理想值<10ms，>50ms说明loop被CPU密集型任务或阻塞IO拖住。
2. Task Queue Length：等待执行的协程数量。持续>100说明并发压力过大或下游响应慢。
3. Connection Pool Usage：数据库/HTTP连接池的使用率。>90%意味着需要扩容。

我们用aiometer库暴露metrics：

# metrics.py from aiometer import run_all import asyncio async def monitor_event_loop_lag(): loop = asyncio.get_running_loop() # 测量loop处理一个空任务的耗时 start = loop.time() await asyncio.sleep(0) # 让出一次控制权 lag = loop.time() - start return lag # 在FastAPI中间件中定期采集 @app.middleware("http") async def add_metrics(request: Request, call_next): start_time = asyncio.get_event_loop().time() response = await call_next(request) process_time = asyncio.get_event_loop().time() - start_time # 上报process_time, status_code等 return response

Grafana面板里，我们设置告警规则：当event_loop_lag_seconds{job="weather-api"} > 0.05持续2分钟，立即触发PagerDuty。上线后第一次告警，我们发现是某个cpu_heavy_calculation没走run_in_executor，修复后lag回归正常。这就是异步监控的价值：它不告诉你“服务挂了”，而是告诉你“哪里卡住了”，让你精准定位。

5. 常见问题与排查技巧实录：那些文档里不会写的血泪教训

5.1 “RuntimeWarning: coroutine 'xxx' was never awaited”：不是警告，是内存泄漏炸弹

这个警告看似温和，实则是严重bug。它意味着你创建了一个协程对象，但从未await或asyncio.run()它。Python不会自动执行，也不会垃圾回收——因为协程对象持有对所有局部变量的引用，导致这些变量永远无法被释放。

真实案例：我们有个定时任务函数：

# ❌ 危险代码 async def cleanup_old_data(): await db.execute("DELETE FROM logs WHERE created_at < NOW() - INTERVAL '7 days'") # 在同步cron job里调用 def cron_job(): cleanup_old_data() # 忘了await！只创建协程对象，不执行

结果：每天生成12个未执行的协程对象，每个持有一个数据库连接引用，7天后内存暴涨1.2GB，OOM Killer干掉了进程。修复方案只有两个：

• 在同步代码中，必须用asyncio.run(cleanup_old_data())
• 或者，把cron job本身改成异步（用asyncio.run()包装）

提示：在CI/CD流水线中加入pylint检查，规则W0104（statement has no effect）能捕获这种async def调用未await的情况。

5.2 “Task was destroyed but it is pending!”：未完成的Task正在偷偷吃掉你的连接

当event loop关闭时（如Uvicorn重启），所有未完成的Task会被强制取消。如果这些Task正在执行await httpx.AsyncClient.get()，而HTTP连接还没关闭，就会出现这个警告。更糟的是，这些“幽灵连接”会一直挂在操作系统层面，直到TCP超时（通常几分钟），耗尽你的文件描述符（fd）。

排查步骤：

1. lsof -i -n -p <pid>查看进程打开的socket，过滤TIME_WAIT状态
2. 如果数量>1000，基本确认是未清理的Task
3. 在on_shutdown事件中，显式等待所有Task完成：

@app.on_event("shutdown") async def shutdown_event(): # 取消所有还在运行的Task tasks = [t for t in asyncio.all_tasks() if t is not asyncio.current_task()] [t.cancel() for t in tasks] # 等待它们完成（最多5秒） await asyncio.gather(*tasks, return_exceptions=True) # 关闭资源 await app.state.weather_dal.pool.close() await app.state.api_client.client.aclose()

5.3 “SSL handshake failed”：异步HTTP客户端的证书信任链陷阱

httpx默认使用系统证书，但某些Linux发行版（如Alpine）的ca-certificates包不完整。aiohttp会静默失败，而httpx直接抛httpx.ConnectError。解决方案不是禁用SSL验证（绝对禁止！），而是显式指定证书路径：

import ssl import httpx # 从系统加载证书 context = ssl.create_default_context() context.load_verify_locations("/etc/ssl/certs/ca-certificates.crt") client = httpx.AsyncClient(verify=context)

或者，在Dockerfile中确保安装完整证书：

FROM python:3.11-slim RUN apt-get update && apt-get install -y ca-certificates && rm -rf /var/lib/apt/lists/* COPY . /app WORKDIR /app

5.4 性能倒退：为什么加了async，QPS反而下降了30%？

这是最打击信心的问题。我们遇到过三次，原因各不相同：

实操心得：异步不是银弹。它放大了所有底层问题。在优化async之前，先用asyncpg的EXPLAIN ANALYZE查SQL，用httpx的log_level="DEBUG"看HTTP连接复用率，用asyncio的loop.slow_callback_duration=0.01找出耗时>10ms的回调——这才是高手的调试路径。

5.5 迁移路线图：如何把一个10万行的Django项目渐进式异步化？

没人能一夜之间把整个Django项目async化。我们的路线图分四步，每步可独立上线：

1. Step 1：异步视图（Django 4.1+）
   把高IO、低计算的视图（如调用外部API）改为async def，用sync_to_async()包装数据库操作。

   from asgiref.sync import sync_to_async @sync_to_async def sync_db_query(user_id): return User.objects.get(id=user_id) async def async_view(request): user = await sync_db_query(123) # 安全！ return JsonResponse({"user":