FastAPI GPU推理服务并发控制实战：从显存溢出到四层防线

1. 从一次线上事故说起为什么并发控制是GPU推理服务的生死线去年冬天我帮一个团队收拾过一个烂摊子。他们用 FastAPI 包了一个视觉推理模型部署到一台单卡 24G 显存的机器上接口压测的时候一切正常单请求延迟 80ms 左右QPS 跑到 30 都没问题。结果上线第二天业务方搞了个批量任务瞬间打进来两百多个并发请求服务在 3 秒内直接挂掉日志里清一色是CUDA out of memory进程被系统 OOM Killer 干掉连带着整台机器上的其他服务一起遭殃。这个场景太典型了。FastAPI 本身是异步框架天生适合处理高并发 IO但 GPU 推理是典型的计算密集型 独占资源型任务两者放在一起如果不做任何控制异步框架的“来者不拒”会直接把 GPU 显存冲爆。问题的本质不是 FastAPI 不好也不是模型太大而是请求的并发度远远超过了 GPU 实际能承载的并发度。这篇文章我想把这件事彻底讲透。核心围绕三个问题展开第一GPU 推理的并发瓶颈到底在哪里为什么显存会溢出第二FastAPI 里有哪些并发控制手段各自适合什么场景第三怎么把这些手段组合成一套能扛住真实流量的方案。内容适合已经用 FastAPI 部署过模型、但被并发问题折磨过的同学也适合正准备把本地推理脚本包装成线上服务的开发者。读完你应该能直接照着搭一套带并发控制的推理服务而不是等线上炸了再回来补。2. 先搞清楚敌人GPU 推理的并发瓶颈到底在哪2.1 显存溢出的三种典型触发路径很多人以为显存溢出就是“模型太大装不下”其实线上事故里模型本身占的那部分显存往往是固定的真正把显存吃爆的是动态部分。我把它拆成三条路径你对号入座。第一条是请求并发导致的激活值堆积。深度学习推理时每一层的前向计算都会产生中间激活张量这些张量在计算完成前必须留在显存里。单个请求的激活值可能只有几百 MB但十个请求同时进来激活值就是十倍。更麻烦的是PyTorch 的 CUDA 缓存分配器不会立刻把释放的显存还给系统而是留在缓存池里备用所以你看到的nvidia-smi显存占用会一直维持在高位直到缓存被复用或者手动清理。第二条是批处理尺寸失控。有些同学为了提升吞吐会把多个请求攒成一个 batch 一起推理。这个思路本身没错但如果没有上限业务高峰期 batch 会越攒越大显存需求线性增长最后直接爆掉。我见过一个服务batch 攒到 64 的时候显存占用从 4G 飙到 22G就是因为没设 batch 上限。第三条是模型副本和上下文泄漏。比如每个请求都重新加载一次模型或者推理过程中创建的临时 tensor 没有被正确释放又或者用了torch.no_grad()之外的上下文导致计算图被保留。这类问题最隐蔽因为它在低并发下完全看不出来只有压力上来才暴露。2.2 为什么 FastAPI 的异步模型会放大这个问题FastAPI 基于 Starlette底层是 asyncio 事件循环。它的设计哲学是一个 worker 进程可以同时处理成百上千个连接只要这些连接的处理逻辑是异步非阻塞的。问题在于GPU 推理是阻塞的。当你写result model(input)这一行的时候Python 主线程会一直卡在那里等 CUDA kernel 执行完事件循环被阻塞其他请求只能排队。这时候有两种写法。一种是直接在async def里调用同步的推理函数这会导致事件循环被阻塞并发能力反而比同步框架还差。另一种是用run_in_executor把推理丢到线程池里这样事件循环不被阻塞但线程池的并发数如果设得太大就会有大量请求同时抢 GPU显存瞬间爆炸。所以 FastAPI 的异步特性在这里是一把双刃剑它让请求接收变得极其高效但如果没有在推理这一层做限流请求会像洪水一样涌向 GPU。并发控制的核心就是在请求接收和 GPU 执行之间加一道可控的闸门。2.3 一个必须记住的容量估算公式在动手写代码之前先算一笔账。假设你的模型权重占WGB单个请求的峰值激活值占AGBCUDA 上下文和框架本身占CGB通常 0.5 到 1.5 GB那么单卡能安全承载的并发请求数N大致满足N (Total_VRAM - W - C) / A举个例子24G 显存的卡模型权重 6G框架开销 1G单请求激活值 0.8G那么N (24 - 6 - 1) / 0.8 21.25取整就是 21。但这是理论极限实际要留 20% 到 30% 的余量应对显存碎片和突发峰值所以安全并发数应该设在 14 到 16 之间。这个公式看起来简单但A这个值很多人不知道怎么测。我的做法是写一个脚本用torch.cuda.max_memory_allocated()在单请求推理前后打点跑十次取最大值。注意要用max_memory_allocated而不是memory_allocated因为前者记录的是峰值后者只是当前值。测出来之后再乘以一个 1.3 的安全系数就是你在代码里应该用的单请求显存预算。3. FastAPI 并发控制的四层防线从入口到 GPU 逐层拦截3.1 第一层信号量控制推理并发数最直接的手段是用asyncio.Semaphore限制同时进入推理的请求数。信号量的值就设成上面算出来的安全并发数。写法很简单import asyncio from fastapi import FastAPI app FastAPI() GPU_SEMAPHORE asyncio.Semaphore(14) app.post(/infer) async def infer(payload: dict): async with GPU_SEMAPHORE: result await run_inference(payload) return result这段代码的关键在于async with的位置。信号量必须在进入推理之前获取在推理完成之后释放。如果你把run_inference写成同步函数记得用loop.run_in_executor包一层否则信号量虽然限制了并发数但事件循环还是会被阻塞。这里有个坑我踩过asyncio.Semaphore是绑定事件循环的如果你用多 worker 启动比如uvicorn --workers 4每个 worker 进程有自己独立的事件循环和信号量实际并发数会变成14 * 4 56。所以要么用单 worker 加多线程要么把信号量的值除以 worker 数。我个人的建议是推理服务用单 worker因为 GPU 本来就是独占资源多 worker 只会让显存管理更复杂。3.2 第二层请求队列与背压机制信号量能限制并发但限制不了排队。当 200 个请求同时进来14 个在执行剩下 186 个在等信号量。如果这些请求一直堆积内存会涨客户端会超时体验很差。这时候需要背压当队列长度超过阈值时直接拒绝新请求返回 503。from fastapi import HTTPException import asyncio MAX_QUEUE 50 queue_size 0 app.post(/infer) async def infer(payload: dict): global queue_size if queue_size MAX_QUEUE: raise HTTPException(status_code503, detailserver busy) queue_size 1 try: async with GPU_SEMAPHORE: result await run_inference(payload) return result finally: queue_size - 1MAX_QUEUE怎么定我的经验是设成并发数 * 3到并发数 * 5。比如并发 14队列设 50 左右。太小会导致正常突发流量被误拒太大则排队时间过长客户端早就超时了。另外记得在finally里减计数否则异常路径会导致计数泄漏队列越来越小。注意queue_size这种全局变量在单 worker 下没问题但如果你用了多进程或者多线程必须换成multiprocessing.Value或者加锁否则计数会错乱。3.3 第三层动态批处理用吞吐换延迟如果你的业务对延迟不那么敏感但对吞吐要求高动态批处理是绕不开的。思路是收集一小段时间窗口内的请求凑成一个 batch 一起送进 GPU这样 GPU 利用率更高单位显存的吞吐更大。实现上可以用一个后台任务维护一个待处理队列每隔batch_timeout毫秒或者队列长度达到max_batch_size就触发一次推理。FastAPI 这边用asyncio.Future把结果回传给各个请求。import asyncio from collections import deque BATCH_TIMEOUT 0.02 # 20ms MAX_BATCH_SIZE 8 pending deque() batch_lock asyncio.Lock() async def batch_worker(): while True: await asyncio.sleep(BATCH_TIMEOUT) async with batch_lock: if not pending: continue batch [] while pending and len(batch) MAX_BATCH_SIZE: batch.append(pending.popleft()) inputs [item[0] for item in batch] try: outputs await run_batch_inference(inputs) for (_, future), out in zip(batch, outputs): if not future.done(): future.set_result(out) except Exception as e: for _, future in batch: if not future.done(): future.set_exception(e)这个方案的关键参数是BATCH_TIMEOUT和MAX_BATCH_SIZE。BATCH_TIMEOUT越大攒的 batch 越大吞吐越高但延迟也越高。20ms 是个比较平衡的值对大多数业务来说用户感知不到。MAX_BATCH_SIZE则要根据显存预算来定用前面那个公式反推max_batch_size (Total_VRAM - W - C) / A。动态批处理有个隐藏收益它天然实现了并发控制。因为 batch 大小有上限同时进入 GPU 的请求数就被限制住了不会出现显存溢出。但它也有代价就是实现复杂度高而且对变长输入比如不同长度的文本需要做 paddingpadding 本身也占显存。3.4 第四层显存监控与熔断降级前面三层都是预防这一层是兜底。即使做了并发控制也可能因为显存碎片、模型异常、输入尺寸超预期等原因导致显存紧张。这时候需要主动监控在溢出之前熔断。import torch def check_gpu_memory(threshold0.9): if not torch.cuda.is_available(): return True allocated torch.cuda.memory_allocated() total torch.cuda.get_device_properties(0).total_memory return (allocated / total) threshold app.post(/infer) async def infer(payload: dict): if not check_gpu_memory(threshold0.9): raise HTTPException(status_code503, detailgpu memory critical) ...阈值设 0.9 是个经验值。超过 90% 之后新的显存分配请求失败概率会显著上升因为碎片化严重。这时候拒绝新请求让已有请求跑完显存会自然回落。另外可以配合torch.cuda.empty_cache()在低峰期主动清理缓存但注意这个操作会带来几十毫秒的卡顿不要在高峰期调用。4. 完整可复现的实战方案从零搭一个带并发控制的推理服务4.1 项目结构与依赖清单我习惯把推理服务拆成几个模块避免所有代码堆在一个main.py里。目录结构大概是这样infer-service/ ├── app/ │ ├── __init__.py │ ├── main.py # FastAPI 入口 │ ├── config.py # 配置参数 │ ├── model.py # 模型加载与推理封装 │ ├── concurrency.py # 并发控制逻辑 │ └── schemas.py # 请求响应模型 ├── requirements.txt └── run.sh依赖清单要锁版本尤其是 PyTorch 和 CUDA 的对应关系版本不匹配会导致各种诡异问题fastapi0.110.0 uvicorn0.29.0 torch2.2.0 pydantic2.6.0 numpy1.26.0run.sh里启动命令用单 workeruvicorn app.main:app --host 0.0.0.0 --port 8000 --workers 1单 worker 的原因前面说过GPU 是独占资源多 worker 只会让显存管理失控。如果单 worker 的 CPU 处理能力不够比如请求体解析很重可以把推理部分放到线程池但 worker 数保持 1。4.2 模型加载只加载一次全局共享模型加载必须放在应用启动时而不是每个请求里。用 FastAPI 的lifespan机制from contextlib import asynccontextmanager from fastapi import FastAPI import torch model None asynccontextmanager async def lifespan(app: FastAPI): global model model load_model() model.eval() if torch.cuda.is_available(): model model.cuda() yield # 清理 del model torch.cuda.empty_cache() app FastAPI(lifespanlifespan)这里有几个细节。model.eval()必须调用否则 dropout 和 batchnorm 会按训练模式走结果不稳定还费显存。torch.no_grad()在推理函数里加不要在这里加因为它是上下文管理器作用范围要精确控制。另外lifespan是 FastAPI 新版本推荐的写法老版本用app.on_event(startup)但那个已经废弃了新项目别再用。4.3 推理函数线程池 信号量的正确组合推理函数是同步的必须丢到线程池否则阻塞事件循环。但线程池的大小要和信号量配合import asyncio from concurrent.futures import ThreadPoolExecutor import torch GPU_CONCURRENCY 14 executor ThreadPoolExecutor(max_workersGPU_CONCURRENCY) semaphore asyncio.Semaphore(GPU_CONCURRENCY) def _sync_infer(input_tensor): with torch.no_grad(): input_tensor input_tensor.cuda() output model(input_tensor) result output.cpu().numpy() return result async def run_inference(payload): loop asyncio.get_running_loop() tensor preprocess(payload) async with semaphore: result await loop.run_in_executor(executor, _sync_infer, tensor) return postprocess(result)线程池大小和信号量值设成一样这样不会出现“信号量放行了但线程池没位置”的情况。torch.no_grad()放在_sync_infer里确保每次推理都不建计算图。input_tensor.cuda()和output.cpu()的搬运是必须的因为预处理和后处理都在 CPU 上做只有模型计算在 GPU 上。有个性能细节如果输入 tensor 很小.cuda()的搬运开销可能比计算本身还大。这种情况下可以考虑用pin_memory加速或者把预处理也放到 GPU 上。但后者会让显存占用增加要重新算并发数。4.4 参数计算用实测数据反推并发数前面给的公式是估算实际部署前一定要实测。我写了一个压测脚本用locust或者简单的asyncio并发请求逐步增加并发数观察显存和延迟import asyncio import aiohttp import torch async def single_request(session, url, payload): async with session.post(url, jsonpayload) as resp: return await resp.json() async def stress_test(concurrency): url http://localhost:8000/infer payload {input: test} async with aiohttp.ClientSession() as session: tasks [single_request(session, url, payload) for _ in range(concurrency)] results await asyncio.gather(*tasks, return_exceptionsTrue) return results async def main(): for c in [1, 2, 4, 8, 16, 32]: torch.cuda.reset_peak_memory_stats() results await stress_test(c) peak torch.cuda.max_memory_allocated() / 1024**3 print(fconcurrency{c}, peak_vram{peak:.2f}GB)跑完这个脚本你会得到一张并发数 vs 峰值显存的表。找到显存占用接近总量 80% 的那个并发数就是你的安全并发上限。我实测过一个 ResNet50 的服务并发 16 的时候峰值显存 18.2G并发 32 直接 OOM所以最终把信号量设在了 14。5. 踩坑实录那些文档里不会写的并发控制问题5.1 信号量泄漏导致服务假死有一次服务跑了两天之后突然所有请求都超时但nvidia-smi显示显存占用很低GPU 利用率也是 0。排查了半天才发现是信号量泄漏某个请求在获取信号量之后抛了异常但异常处理路径里没有释放信号量导致信号量计数永远减不回去后续请求全部卡在async with上。解决办法是用try/finally包住信号量释放或者直接用async with因为async with本身保证了异常时也会释放。但如果你手动调用了acquire()就必须手动release()而且要放在finally里。这个坑我踩过一次之后现在所有信号量操作都用async with绝不用手动 acquire。5.2 CUDA 缓存导致的“假性显存溢出”PyTorch 的 CUDA 缓存分配器有个特性它释放显存时不会立刻还给系统而是留在缓存池里。这导致nvidia-smi看到的显存占用总是偏高但实际上这部分显存是可复用的。如果你用nvidia-smi的数值来判断是否溢出会误判。正确的判断方式是看torch.cuda.memory_allocated()实际被 tensor 占用的和torch.cuda.memory_reserved()缓存池占用的。如果allocated很低但reserved很高说明是缓存问题不是真溢出。这时候可以调torch.cuda.empty_cache()释放缓存但注意这个操作会同步等待所有 CUDA 操作完成有卡顿。5.3 多线程下的 GIL 与 CUDA 上下文竞争用ThreadPoolExecutor跑推理时多个线程会共享同一个 CUDA 上下文。CUDA 上下文本身是线程安全的但 PyTorch 的某些操作在释放 GIL 之后可能会有竞争。我遇到过的情况是并发 14 的时候偶尔会出现某个请求的推理结果错乱返回了另一个请求的输出。排查后发现是输入 tensor 的复用问题预处理函数里用了一个全局的 numpy buffer多线程同时写导致数据串了。解决办法是每次预处理都新建 tensor不要复用全局 buffer。这个坑很隐蔽因为低并发下几乎不会触发只有压力上来才偶发。5.4 常见问题速查表现象可能原因排查方法解决手段CUDA out of memory并发数超限看max_memory_allocated降低信号量值请求全部超时但 GPU 空闲信号量泄漏打印信号量计数用async with替代手动 acquire显存占用高但利用率低CUDA 缓存未释放对比 allocated 和 reserved低峰期调empty_cache推理结果错乱全局 buffer 竞争检查预处理函数每次新建 tensor服务启动后第一个请求特别慢模型懒加载看启动日志用 lifespan 预加载多 worker 下并发数翻倍信号量不跨进程看 worker 数改单 worker6. 进阶思路让并发控制更聪明一点6.1 基于显存水位的自适应并发固定并发数的问题是它假设每个请求的显存开销都一样。但实际业务里输入尺寸可能差异很大一个长文本请求的激活值可能是短文本的十倍。这时候固定并发数要么太保守浪费资源要么太激进溢出。自适应方案是维护一个当前显存水位每次请求进来时检查水位如果低于阈值就放行高于阈值就排队。实现上可以用一个asyncio.Condition配合一个后台任务定期更新水位。import asyncio import torch class AdaptiveLimiter: def __init__(self, high_watermark0.85, low_watermark0.7): self.high high_watermark self.low low_watermark self.condition asyncio.Condition() self.running 0 async def acquire(self): async with self.condition: while self._memory_ratio() self.high: await self.condition.wait() self.running 1 async def release(self): async with self.condition: self.running - 1 self.condition.notify_all() def _memory_ratio(self): if not torch.cuda.is_available(): return 0 return torch.cuda.memory_allocated() / torch.cuda.get_device_properties(0).total_memory这个方案比固定信号量灵活但实现复杂度高而且_memory_ratio的读取有延迟可能在高频请求下判断不准。我的建议是如果业务输入尺寸比较均匀用固定信号量就够了如果差异很大再考虑自适应。6.2 优先级队列让重要请求先跑有些业务场景下不是所有请求都平等。比如付费用户的请求应该优先处理或者某些实时性要求高的接口不能被批量任务堵住。这时候可以用优先级队列替代普通队列。asyncio.PriorityQueue可以做到但要注意它要求元素可比较。通常的做法是封装一个带优先级的对象重写__lt__方法。不过优先级队列有个副作用低优先级请求可能永远排不到导致饥饿。解决办法是给每个请求加一个等待时间权重等待越久优先级越高。6.3 多卡场景下的负载均衡如果机器上有多张 GPU并发控制就要考虑卡间均衡。最简单的做法是每张卡一个信号量请求进来时轮询分配。但要注意不同卡的显存占用可能不一样轮询会导致某张卡先满。更好的做法是每次选择当前显存占用最低的卡。def select_gpu(): if not torch.cuda.is_available(): return None min_mem float(inf) best 0 for i in range(torch.cuda.device_count()): mem torch.cuda.memory_allocated(i) if mem min_mem: min_mem mem best i return best多卡场景下模型要么每张卡一份副本显存翻倍要么用模型并行实现复杂。大多数情况下每卡一份副本加请求级负载均衡是最实用的方案。7. 我个人的几条实战心得第一并发数宁小勿大。我见过太多人为了追求吞吐把并发数设得很激进结果线上稍微有点波动就 OOM。显存溢出是灾难性的整个服务挂掉而并发数小一点只是延迟高一点用户还能用。留 30% 的显存余量是底线。第二压测一定要用真实数据。用{input: test}这种假数据压测测出来的显存占用和真实业务差很远。真实数据的长度分布、batch 组成都会影响显存。我一般会从线上日志里采样一批真实请求做压测。第三监控比控制更重要。并发控制是预防但你不能保证预防一定成功。所以一定要有显存监控和告警在溢出之前就能发现趋势。我习惯在服务里暴露一个/metrics接口输出当前显存占用、并发数、队列长度接到监控系统里。第四别迷信框架自带的限流。FastAPI 本身没有内置的并发控制slowapi之类的限流库是针对请求频率的不是针对 GPU 资源的。GPU 并发控制必须自己实现因为只有你知道模型的显存开销。第五优雅降级比硬扛更重要。当显存紧张时与其让请求全部卡死不如主动拒绝一部分返回 503 让客户端重试。配合客户端的重试策略整体可用性反而更高。这个思路在分布式系统里叫“快速失败”在 GPU 推理服务里同样适用。这套方案我在三个不同的推理服务上落地过从单卡 3090 到多卡 A100 集群核心逻辑都是一样的算清楚容量设好闸门监控水位留好退路。GPU 推理服务的稳定性本质上不是模型问题是资源管理问题。把并发控制做扎实比换更大的卡更有效。