拓冰网站定制 
本文摘要压测里 QPS 上不去、P99 突刺、健康检查超时多因async def中发同步请求挡住事件循环。用 1/N 算术与循环停顿探针定位阻塞点再比较def、线程池与异步客户端的修复代价。一、问题与结论压测配置uvicorn app:app单 worker、本地慢上游DELAY 0.2、并发 8、每条路由 32 个请求。/bad的 RPS 约为/ok-*的 1/81/N 推导值实测以脚本为准尾延迟与/health最大耗时同步抬高并发改回 1四条路由几乎一样快。根因是async def路由里发了同步请求urllib.request.urlopen、requests、同步 DB 驱动都算。同步调用期间协程不让出控制权事件循环整体停摆N 个并发被串行成 1 个1/8 不是框架系数是并发数 8 的算术结果。二、排查与选择依据先分清慢在上游还是慢在自己单请求冒烟正常、并发一上就塌陷瓶颈就在进程内调度。三条互证手段成本从低到高loop-lag探针启动时跑一个asyncio.sleep(0.2)的心跳任务测量实际唤醒延迟。压测/bad时打印200ms量级的停顿、压测/ok-*时不打印就是循环被占住的直接证据。健康检查波及/health与业务路由共用事件循环业务阻塞时它同样排不上队压测/bad期间health_max抬高意味着 K8s liveness 探针可能超时、Pod 反复重启。py-spy dump --pid pid改成def后若高并发 P99 反而更差抓线程栈大量线程停在socket read说明排队点已从事件循环转到线程池该加超时和限流而不是继续加线程。选型看三件事调用能否换成异步客户端、同步 SDK 是否线程安全、并发量是否超过线程槽位。替代方案与取舍方案选择条件代价边界何时不该用路由改声明def阻塞调用不可替换、想改一行受框架线程池上限约束线程有内存与切换成本上游无超时时会占满槽位高并发 P99 更差run_in_threadpool/asyncio.to_thread保留async def结构只挪一段调用两者线程池不同、上限不互通需自行限流同步对象跨线程共享前必须确认线程安全换httpx.AsyncClient调用面可重写、并发量高连接池、超时、重试都要显式配置只有同步 SDK、或同步依赖过深时重构成本过高增加--workers需要快速止血内存与线程数翻倍掩盖根因不能当作修复尾延迟与探针抖动依旧进程外执行队列/独立服务慢 I/O 需与在线链路解耦序列化、失败重试、可观测性成本只适合可异步化、可延迟的调用纯 CPU 密集路由不该用任何线程方案线程只增加切换开销应放进程池或独立服务低流量接口也看不出差别不必重构。无论选哪种方案入口并发限流与上游强制超时都要配它们不提速但能防止慢上游占满线程槽位。三、关键原理async def路由是协程直接跑在事件循环上def路由交给框架线程池执行后再await。协程在await之前不让出控制权一次同步阻塞期间同一循环上的所有 ready 回调其他请求、超时计时、健康检查都被延后延后量约等于阻塞耗时。吞吐按 1/N 坍塌是算术推导不是实测数字。设上游延迟为 T、并发为 N写法每请求耗时N 并发下的吞吐真异步awaitTN 个重叠N/Tasync def里同步阻塞串行为 N·T1/T比值 1/N并发 8 就是 1/8并发 32 就是 1/32单请求测试完全看不出差别。适用条件是上游 I/O 等待占主导、CPU 开销可忽略、单事件循环CPU 密集或连接池竞争时比值会偏离。多开 M 个 worker 时比值约为 1/⌈N/M⌉但每个循环内部仍在停顿。线程池归属是常被忽略的成本路由写def与run_in_threadpool走框架侧线程池受 anyio 的 thread limiter 限制asyncio.to_thread与loop.run_in_executor(None, ...)走事件循环默认的ThreadPoolExecutor。两者上限互不相通一处修好、另一处排队很常见。默认 token 数、max_workers、连接池与超时随版本变化需按安装版本核实本文不写死数值。四、可运行示例环境Python 3 pip install fastapi uvicorn httpx。app.py内置本地慢上游不依赖外网DELAY可调。# app.pyimportasyncioimportthreadingimporttimefromcontextlibimportasynccontextmanagerfromhttp.serverimportBaseHTTPRequestHandler,ThreadingHTTPServerfromurllib.requestimporturlopenimporthttpxfromfastapiimportFastAPIfromstarlette.concurrencyimportrun_in_threadpool UPSTREAMhttp://127.0.0.1:9100/slowDELAY0.2classSlowHandler(BaseHTTPRequestHandler):defdo_GET(self):time.sleep(DELAY)# 只在上游线程里 sleepbodyb{ok:true}self.send_response(200)self.send_header(Content-Type,application/json)self.send_header(Content-Length,str(len(body)))self.end_headers()self.wfile.write(body)deflog_message(self,*args):passdeffetch_sync()-bytes:# 同步阻塞调用withurlopen(UPSTREAM,timeout5)asr:returnr.read()asyncdefloop_lag_probe():# 事件循环停顿探针loopasyncio.get_running_loop()whileTrue:t0loop.time()awaitasyncio.sleep(0.2)lag(loop.time()-t0)-0.2iflag0.05:print(f[loop-lag] {lag*1000:.0f}ms,flushTrue)asynccontextmanagerasyncdeflifespan(app):srvThreadingHTTPServer((127.0.0.1,9100),SlowHandler)threading.Thread(targetsrv.serve_forever,daemonTrue).start()app.state.clienthttpx.AsyncClient(timeout5)# 生命周期内复用probeasyncio.create_task(loop_lag_probe())yieldprobe.cancel()awaitasyncio.gather(probe,return_exceptionsTrue)# 取消并回收探针任务awaitapp.state.client.aclose()srv.shutdown()appFastAPI(lifespanlifespan)app.get(/bad)asyncdefbad():# 问题写法async def 同步 I/Oreturnfetch_sync()app.get(/ok-threadpool)asyncdefok_threadpool():# 修法 A显式挪进线程池returnawaitrun_in_threadpool(fetch_sync)app.get(/ok-def)defok_def():# 修法 B声明 def交给框架线程池returnfetch_sync()app.get(/ok-async)asyncdefok_async():# 修法 C换异步客户端return(awaitapp.state.client.get(UPSTREAM)).json()app.get(/health)asyncdefhealth():# 观察循环停顿是否波及健康检查return{ok:True}# load.pyimportasyncioimporttimeimporthttpx BASEhttp://127.0.0.1:8000asyncdefrun(path,concurrency8,total32):lat,health,stop[],[],asyncio.Event()asyncdefone(client):t0time.perf_counter()awaitclient.get(BASEpath)returntime.perf_counter()-t0asyncdefhealth_probe(client):whilenotstop.is_set():t0time.perf_counter()awaitclient.get(BASE/health)health.append(time.perf_counter()-t0)awaitasyncio.sleep(0.2)asyncwithhttpx.AsyncClient(timeout30)asclient:probeasyncio.create_task(health_probe(client))semasyncio.Semaphore(concurrency)asyncdefguarded():asyncwithsem:lat.append(awaitone(client))t0time.perf_counter()awaitasyncio.gather(*[guarded()for_inrange(total)])walltime.perf_counter()-t0 stop.set()awaitprobe lat.sort()qlambdap:lat[min(len(lat)-1,int(len(lat)*p))]*1000print(f{path}: RPS{total/wall:.1f}p50{q(.5):.0f}ms p95{q(.95):.0f}ms fp99{q(.99):.0f}ms health_max{max(health)*1000:.0f}ms)if__name____main__:forpin[/bad,/ok-threadpool,/ok-def,/ok-async]:asyncio.run(run(p,concurrency8,total32))运行uvicorn app:app--port8000python load.py预期输出四行含RPS、p50、p95、p99、health_max的结果。并发 8 时/bad的 RPS 应约为/ok-*的 1/8health_max明显更高服务端只在压测/bad期间打印[loop-lag]。这是机制推导的预期形态具体数值未验证以本机实测为准。实际输出数字由本机跑出后填入本文不给实测值。做证伪验证把load.py的concurrency依次改成 1、4、16若/bad与/ok-*的 RPS 比值随并发近似 1/N 变化即坐实串行化若比值不随并发变化应转向上游或连接池排查。常见失败一改成def就上线、没给上游调用加超时。慢上游逐个占满线程槽位低并发变快、高并发 P99 反而更糟网关回 503/504py-spy dump --pid pid可见大量线程停在socket read。修复为上游设短超时、入口加限流再决定是否调大线程 limiter。常见失败二/ok-async每请求新建AsyncClient连接无法复用握手与TIME_WAIT开销在压测下放大未实测。修复像示例一样在生命周期里持有一个客户端实例并显式配置连接池与超时。五、验证结果与边界适用范围是 I/O 密集、外部 HTTP 调用占主导的 FastAPI 服务同步调用挪出事件循环或换成异步客户端后理论上可取回约 N 倍并发能力算术推导未实测。结论建立在单事件循环、阻塞 I/O 占主导之上脚本可复现、可证伪。不该用的场景CPU 密集路由线程只增加切换与内存开销应进程外计算只能用同步 SDK 且对象跨线程不安全需每线程独立连接或严格限流。多 worker 后看似正常的服务先确认每个循环是否仍在停顿--workers只是把 1/N 摊薄到 1/⌈N/M⌉。上线前按安装版本核实anyio thread limiter 默认 token 数、ThreadPoolExecutor默认max_workers、uvicorn 默认 worker 数、httpx连接池与默认超时。这些参数随版本与平台变化本文不写死数值。思考线程 limiter 的 token 数应按上游 P95 延迟倒推还是按 CPU 核数给定健康检查与业务共用事件循环时独立端口与彻底消除阻塞调用哪种先做参考资料FastAPI 官方文档 · Concurrency and async / awaitStarlette 官方文档 · Concurrencyrun_in_threadpool 语义Python 官方文档 · asyncio 任务asyncio.to_threadPython 官方文档 · concurrent.futuresThreadPoolExecutor 默认 max_workersAnyIO 官方文档 · Threads默认 thread limiteruvicorn 官方文档 · Settingsworkers 与 loop 设置 在 async def 里发同步请求，FastAPI 吞吐为何只剩 1/8
本文摘要压测里 QPS 上不去、P99 突刺、健康检查超时多因async def中发同步请求挡住事件循环。用 1/N 算术与循环停顿探针定位阻塞点再比较def、线程池与异步客户端的修复代价。一、问题与结论压测配置uvicorn app:app单 worker、本地慢上游DELAY 0.2、并发 8、每条路由 32 个请求。/bad的 RPS 约为/ok-*的 1/81/N 推导值实测以脚本为准尾延迟与/health最大耗时同步抬高并发改回 1四条路由几乎一样快。根因是async def路由里发了同步请求urllib.request.urlopen、requests、同步 DB 驱动都算。同步调用期间协程不让出控制权事件循环整体停摆N 个并发被串行成 1 个1/8 不是框架系数是并发数 8 的算术结果。二、排查与选择依据先分清慢在上游还是慢在自己单请求冒烟正常、并发一上就塌陷瓶颈就在进程内调度。三条互证手段成本从低到高loop-lag探针启动时跑一个asyncio.sleep(0.2)的心跳任务测量实际唤醒延迟。压测/bad时打印200ms量级的停顿、压测/ok-*时不打印就是循环被占住的直接证据。健康检查波及/health与业务路由共用事件循环业务阻塞时它同样排不上队压测/bad期间health_max抬高意味着 K8s liveness 探针可能超时、Pod 反复重启。py-spy dump --pid pid改成def后若高并发 P99 反而更差抓线程栈大量线程停在socket read说明排队点已从事件循环转到线程池该加超时和限流而不是继续加线程。选型看三件事调用能否换成异步客户端、同步 SDK 是否线程安全、并发量是否超过线程槽位。替代方案与取舍方案选择条件代价边界何时不该用路由改声明def阻塞调用不可替换、想改一行受框架线程池上限约束线程有内存与切换成本上游无超时时会占满槽位高并发 P99 更差run_in_threadpool/asyncio.to_thread保留async def结构只挪一段调用两者线程池不同、上限不互通需自行限流同步对象跨线程共享前必须确认线程安全换httpx.AsyncClient调用面可重写、并发量高连接池、超时、重试都要显式配置只有同步 SDK、或同步依赖过深时重构成本过高增加--workers需要快速止血内存与线程数翻倍掩盖根因不能当作修复尾延迟与探针抖动依旧进程外执行队列/独立服务慢 I/O 需与在线链路解耦序列化、失败重试、可观测性成本只适合可异步化、可延迟的调用纯 CPU 密集路由不该用任何线程方案线程只增加切换开销应放进程池或独立服务低流量接口也看不出差别不必重构。无论选哪种方案入口并发限流与上游强制超时都要配它们不提速但能防止慢上游占满线程槽位。三、关键原理async def路由是协程直接跑在事件循环上def路由交给框架线程池执行后再await。协程在await之前不让出控制权一次同步阻塞期间同一循环上的所有 ready 回调其他请求、超时计时、健康检查都被延后延后量约等于阻塞耗时。吞吐按 1/N 坍塌是算术推导不是实测数字。设上游延迟为 T、并发为 N写法每请求耗时N 并发下的吞吐真异步awaitTN 个重叠N/Tasync def里同步阻塞串行为 N·T1/T比值 1/N并发 8 就是 1/8并发 32 就是 1/32单请求测试完全看不出差别。适用条件是上游 I/O 等待占主导、CPU 开销可忽略、单事件循环CPU 密集或连接池竞争时比值会偏离。多开 M 个 worker 时比值约为 1/⌈N/M⌉但每个循环内部仍在停顿。线程池归属是常被忽略的成本路由写def与run_in_threadpool走框架侧线程池受 anyio 的 thread limiter 限制asyncio.to_thread与loop.run_in_executor(None, ...)走事件循环默认的ThreadPoolExecutor。两者上限互不相通一处修好、另一处排队很常见。默认 token 数、max_workers、连接池与超时随版本变化需按安装版本核实本文不写死数值。四、可运行示例环境Python 3 pip install fastapi uvicorn httpx。app.py内置本地慢上游不依赖外网DELAY可调。# app.pyimportasyncioimportthreadingimporttimefromcontextlibimportasynccontextmanagerfromhttp.serverimportBaseHTTPRequestHandler,ThreadingHTTPServerfromurllib.requestimporturlopenimporthttpxfromfastapiimportFastAPIfromstarlette.concurrencyimportrun_in_threadpool UPSTREAMhttp://127.0.0.1:9100/slowDELAY0.2classSlowHandler(BaseHTTPRequestHandler):defdo_GET(self):time.sleep(DELAY)# 只在上游线程里 sleepbodyb{ok:true}self.send_response(200)self.send_header(Content-Type,application/json)self.send_header(Content-Length,str(len(body)))self.end_headers()self.wfile.write(body)deflog_message(self,*args):passdeffetch_sync()-bytes:# 同步阻塞调用withurlopen(UPSTREAM,timeout5)asr:returnr.read()asyncdefloop_lag_probe():# 事件循环停顿探针loopasyncio.get_running_loop()whileTrue:t0loop.time()awaitasyncio.sleep(0.2)lag(loop.time()-t0)-0.2iflag0.05:print(f[loop-lag] {lag*1000:.0f}ms,flushTrue)asynccontextmanagerasyncdeflifespan(app):srvThreadingHTTPServer((127.0.0.1,9100),SlowHandler)threading.Thread(targetsrv.serve_forever,daemonTrue).start()app.state.clienthttpx.AsyncClient(timeout5)# 生命周期内复用probeasyncio.create_task(loop_lag_probe())yieldprobe.cancel()awaitasyncio.gather(probe,return_exceptionsTrue)# 取消并回收探针任务awaitapp.state.client.aclose()srv.shutdown()appFastAPI(lifespanlifespan)app.get(/bad)asyncdefbad():# 问题写法async def 同步 I/Oreturnfetch_sync()app.get(/ok-threadpool)asyncdefok_threadpool():# 修法 A显式挪进线程池returnawaitrun_in_threadpool(fetch_sync)app.get(/ok-def)defok_def():# 修法 B声明 def交给框架线程池returnfetch_sync()app.get(/ok-async)asyncdefok_async():# 修法 C换异步客户端return(awaitapp.state.client.get(UPSTREAM)).json()app.get(/health)asyncdefhealth():# 观察循环停顿是否波及健康检查return{ok:True}# load.pyimportasyncioimporttimeimporthttpx BASEhttp://127.0.0.1:8000asyncdefrun(path,concurrency8,total32):lat,health,stop[],[],asyncio.Event()asyncdefone(client):t0time.perf_counter()awaitclient.get(BASEpath)returntime.perf_counter()-t0asyncdefhealth_probe(client):whilenotstop.is_set():t0time.perf_counter()awaitclient.get(BASE/health)health.append(time.perf_counter()-t0)awaitasyncio.sleep(0.2)asyncwithhttpx.AsyncClient(timeout30)asclient:probeasyncio.create_task(health_probe(client))semasyncio.Semaphore(concurrency)asyncdefguarded():asyncwithsem:lat.append(awaitone(client))t0time.perf_counter()awaitasyncio.gather(*[guarded()for_inrange(total)])walltime.perf_counter()-t0 stop.set()awaitprobe lat.sort()qlambdap:lat[min(len(lat)-1,int(len(lat)*p))]*1000print(f{path}: RPS{total/wall:.1f}p50{q(.5):.0f}ms p95{q(.95):.0f}ms fp99{q(.99):.0f}ms health_max{max(health)*1000:.0f}ms)if__name____main__:forpin[/bad,/ok-threadpool,/ok-def,/ok-async]:asyncio.run(run(p,concurrency8,total32))运行uvicorn app:app--port8000python load.py预期输出四行含RPS、p50、p95、p99、health_max的结果。并发 8 时/bad的 RPS 应约为/ok-*的 1/8health_max明显更高服务端只在压测/bad期间打印[loop-lag]。这是机制推导的预期形态具体数值未验证以本机实测为准。实际输出数字由本机跑出后填入本文不给实测值。做证伪验证把load.py的concurrency依次改成 1、4、16若/bad与/ok-*的 RPS 比值随并发近似 1/N 变化即坐实串行化若比值不随并发变化应转向上游或连接池排查。常见失败一改成def就上线、没给上游调用加超时。慢上游逐个占满线程槽位低并发变快、高并发 P99 反而更糟网关回 503/504py-spy dump --pid pid可见大量线程停在socket read。修复为上游设短超时、入口加限流再决定是否调大线程 limiter。常见失败二/ok-async每请求新建AsyncClient连接无法复用握手与TIME_WAIT开销在压测下放大未实测。修复像示例一样在生命周期里持有一个客户端实例并显式配置连接池与超时。五、验证结果与边界适用范围是 I/O 密集、外部 HTTP 调用占主导的 FastAPI 服务同步调用挪出事件循环或换成异步客户端后理论上可取回约 N 倍并发能力算术推导未实测。结论建立在单事件循环、阻塞 I/O 占主导之上脚本可复现、可证伪。不该用的场景CPU 密集路由线程只增加切换与内存开销应进程外计算只能用同步 SDK 且对象跨线程不安全需每线程独立连接或严格限流。多 worker 后看似正常的服务先确认每个循环是否仍在停顿--workers只是把 1/N 摊薄到 1/⌈N/M⌉。上线前按安装版本核实anyio thread limiter 默认 token 数、ThreadPoolExecutor默认max_workers、uvicorn 默认 worker 数、httpx连接池与默认超时。这些参数随版本与平台变化本文不写死数值。思考线程 limiter 的 token 数应按上游 P95 延迟倒推还是按 CPU 核数给定健康检查与业务共用事件循环时独立端口与彻底消除阻塞调用哪种先做参考资料FastAPI 官方文档 · Concurrency and async / awaitStarlette 官方文档 · Concurrencyrun_in_threadpool 语义Python 官方文档 · asyncio 任务asyncio.to_threadPython 官方文档 · concurrent.futuresThreadPoolExecutor 默认 max_workersAnyIO 官方文档 · Threads默认 thread limiteruvicorn 官方文档 · Settingsworkers 与 loop 设置