FastAPI GPU推理并发控制：解决显存溢出与OOM的实战指南

1. 从一次显存溢出事故说起：FastAPI GPU推理并发控制到底在解决什么问题

去年帮一个团队排查线上问题，场景很典型：他们用FastAPI包了一个视觉推理服务，模型是YOLO系列的检测网络，部署在一张24G显存的卡上。压测的时候单请求跑得好好的，延迟稳定在80毫秒左右，结果一上生产，QPS刚过8，服务就开始报CUDA out of memory，进程直接挂掉，重启之后又挂，循环往复。运维那边看到的现象是"服务隔几分钟就重启一次"，开发这边看到的是"显存莫名其妙就满了"，两边扯了好几天。

这个问题的本质，其实不是模型太大，也不是卡太小，而是并发控制没做。FastAPI默认是异步框架，请求进来之后如果直接丢给GPU去跑，多个请求会同时抢占显存，每个请求都要分配输入张量、中间激活、输出张量，几个请求叠在一起，24G瞬间就见底了。你要知道，推理阶段的显存占用不是线性的，它跟batch size、序列长度、模型结构都有关系，一个请求可能占2G，两个请求可能占5G，三个请求可能就直接爆了。

所以这篇东西想聊的，就是FastAPI + GPU推理场景下，怎么做并发控制才能既不浪费算力，又不把显存撑爆。适合谁看？如果你正在用FastAPI做模型服务化，或者准备把本地的推理脚本包装成HTTP接口对外提供，又或者你已经上线了但被OOM搞得焦头烂额，那这篇内容应该能帮你少走一些弯路。我会从整体设计思路讲起，然后拆解几个核心的并发控制手段，再给出一套可以直接抄的实操方案，最后把我自己踩过的坑和排查技巧整理出来。

需要提前说明的是，并发控制这件事没有银弹，不同的模型、不同的硬件、不同的业务场景，最优解是不一样的。我下面讲的是通用思路加上一些经过验证的具体做法，你拿去之后需要根据自己的实际情况调整参数。

2. 整体设计思路：为什么不能简单粗暴地加锁

2.1 先搞清楚瓶颈到底在哪

很多人一遇到OOM，第一反应是"加个锁吧"，用threading.Lock或者asyncio.Lock把推理部分包起来，保证同一时刻只有一个请求在用GPU。这个做法确实能解决OOM，但代价是吞吐量直接掉到1，GPU利用率可能只有百分之十几，等于花了大价钱买的卡在摸鱼。

所以在动手之前，得先分清楚你的瓶颈是什么。GPU推理的瓶颈通常有三个：

• 显存容量：这是最硬的约束，显存不够就是不够，没有任何办法绕过，只能减少同时驻留的请求数。
• 计算单元利用率：GPU的SM单元是有限的，如果batch size太小，计算单元跑不满，算力就浪费了。
• 数据传输带宽：CPU到GPU的数据拷贝、GPU到CPU的结果回传，这些也是有开销的，尤其是输入数据大的时候。

大部分OOM问题，核心矛盾是显存容量和并发请求数之间的冲突。你要做的是找到一个平衡点：在显存能承受的范围内，尽可能让GPU忙起来。

2.2 三种常见的并发控制策略对比

我整理了一下实际项目中用过的几种策略，各有各的适用场景：

我个人的建议是：先从信号量限流入手，把服务跑稳，然后再根据实际压测数据决定要不要上动态批处理。因为动态批处理的复杂度真的不低，涉及到请求队列管理、超时控制、batch组装、结果拆分分发等一系列问题，如果请求量没到那个级别，投入产出比不划算。

2.3 FastAPI的异步模型对并发控制的影响

这里有个容易被忽略的点：FastAPI是基于asyncio的，它的请求处理是异步的。如果你在async def的路由函数里直接调用同步的GPU推理代码，会阻塞整个事件循环，导致其他请求连进来都进不来。这个行为跟Flask那种多线程模型完全不一样。

所以正确的做法是：把GPU推理放到线程池或者进程池里执行，用run_in_executor或者asyncio.to_thread来调用，这样事件循环不会被阻塞，FastAPI才能正常处理其他请求的接入和排队。

但这里又有一个坑：如果你用线程池，多个线程同时调用CUDA，PyTorch的CUDA上下文是线程安全的，但显存分配是全局的，多个线程同时分配显存还是可能OOM。所以线程池的并发度必须和信号量的并发度配合起来用，不能各管各的。

3. 核心细节解析：信号量、线程池与显存估算

3.1 信号量限流的正确用法

信号量（Semaphore）是控制并发数最直接的工具。在FastAPI里，你可以用asyncio.Semaphore来限制同时执行的推理请求数。基本用法是这样的：

import asyncio from fastapi import FastAPI app = FastAPI() semaphore = asyncio.Semaphore(2) # 最多同时2个推理请求 @app.post("/predict") async def predict(data: dict): async with semaphore: result = await asyncio.to_thread(run_inference, data) return result

这段代码看起来很简单，但有几个细节需要注意。

第一，Semaphore的数值怎么定。这个不能拍脑袋，得根据你的模型显存占用和显卡总显存来算。假设你的模型加载后占8G，每个请求的输入输出和中间激活占3G，显卡是24G，那么理论上最多能同时跑(24 - 8) / 3 = 5个请求。但实际中要留一些余量给CUDA上下文、显存碎片、其他进程等，所以建议取理论值的60%到70%，也就是3个左右。

第二，Semaphore要放在全局还是每个请求创建。必须是全局的，放在模块级别。如果你在路由函数里创建Semaphore，那每个请求都会有自己的Semaphore，完全起不到限流作用。

第三，acquire和release要配对。用async with是最安全的写法，即使推理过程中抛异常，Semaphore也会被正确释放。如果你手动调用acquire()和release()，一定要用try/finally包起来，否则一旦异常，Semaphore就永远不会释放，后续请求全部卡死。

3.2 线程池的配置与陷阱

前面说了，GPU推理要放到线程池里跑，避免阻塞事件循环。asyncio.to_thread底层用的是默认的ThreadPoolExecutor，默认最大线程数是min(32, os.cpu_count() + 4)。这个数值对于GPU推理场景来说通常太大了，因为你的瓶颈在GPU不在CPU，线程开多了反而会增加显存竞争。

我一般会显式创建一个ThreadPoolExecutor，把max_workers设置成和Semaphore一样的值：

from concurrent.futures import ThreadPoolExecutor import asyncio executor = ThreadPoolExecutor(max_workers=2) @app.post("/predict") async def predict(data: dict): async with semaphore: loop = asyncio.get_event_loop() result = await loop.run_in_executor(executor, run_inference, data) return result

这里有个坑：ThreadPoolExecutor的max_workers和Semaphore的数值如果不一致，会出现线程等待或者显存竞争。如果线程数大于Semaphore数，多出来的线程会卡在Semaphore上，浪费资源；如果线程数小于Semaphore数，Semaphore还没满，但线程已经用完了，请求会排队等线程，实际并发度上不去。所以这两个值最好保持一致。

还有一个更隐蔽的坑：PyTorch的CUDA操作在某些情况下不是完全线程安全的。虽然PyTorch官方说CUDA上下文是线程安全的，但在多线程环境下，如果多个线程同时做显存分配和释放，可能会遇到显存碎片问题。表现就是明明总显存够，但就是分配不出来。解决办法是尽量让每个线程的显存分配模式一致，或者用torch.cuda.empty_cache()定期清理，但这个操作本身也有开销，不能频繁调用。

3.3 显存占用的估算方法

要合理设置并发数，你得知道每个请求大概占多少显存。最准确的方法当然是实测，但在上线之前，你可以用下面的方法做个粗略估算。

对于推理阶段，显存占用主要来自这几块：

• 模型参数：这部分是固定的，加载后就占着。可以用torch.cuda.memory_allocated()查看。
• 输入张量：跟输入数据的shape有关。比如一张640x640x3的图片，float32的话就是640 * 640 * 3 * 4 = 4.9MB，不大。
• 中间激活：这部分是大头，跟模型结构有关。对于CNN来说，中间激活的显存占用可能比模型参数还大。
• 输出张量：通常不大，但如果输出是分割掩码或者高分辨率特征图，也会占不少。
• CUDA上下文和显存碎片：这部分通常占1G到2G，不可忽略。

实测的方法很简单：加载模型，跑一个请求，记录torch.cuda.max_memory_allocated()，然后再跑一个并发请求，再记录，两个值的差就是每个请求的增量显存占用。用这个增量去算最大并发数比较靠谱。

注意：torch.cuda.max_memory_allocated()返回的是峰值显存，不是当前显存。如果你想看当前显存，用torch.cuda.memory_allocated()。另外，PyTorch的显存缓存机制会导致memory_allocated和nvidia-smi看到的数值不一致，这是正常的，以nvidia-smi为准。

4. 实操过程：从零搭建一个带并发控制的FastAPI推理服务

4.1 项目目录结构

先说一下目录结构，这个是我自己常用的组织方式，比较清晰：

fastapi-gpu-inference/ ├── app/ │ ├── __init__.py │ ├── main.py # FastAPI入口 │ ├── config.py # 配置参数 │ ├── model.py # 模型加载和推理逻辑 │ ├── schemas.py # 请求/响应数据结构 │ └── concurrency.py # 并发控制相关 ├── requirements.txt └── run.sh

把并发控制单独抽一个模块出来，是为了方便调整参数和复用。模型加载和推理逻辑也单独放，这样换模型的时候不用动主逻辑。

4.2 模型加载与预热

模型加载这块有个关键点：一定要在服务启动的时候加载，不要在请求里加载。我见过有人在路由函数里torch.load()模型，每个请求加载一次，那显存不爆才怪。

# app/model.py import torch import torchvision.models as models class InferenceModel: def __init__(self, device="cuda"): self.device = device self.model = None def load(self): self.model = models.resnet50(pretrained=True) self.model.to(self.device) self.model.eval() # 预热：跑一次空推理，让CUDA上下文和显存分配器初始化 dummy = torch.randn(1, 3, 224, 224).to(self.device) with torch.no_grad(): self.model(dummy) torch.cuda.synchronize() @torch.no_grad() def predict(self, input_tensor): input_tensor = input_tensor.to(self.device) output = self.model(input_tensor) return output.cpu()

预热这一步很多人会省略，但它其实很重要。第一次推理的时候，CUDA会做很多初始化工作，包括编译kernel、分配显存池等，耗时可能是正常推理的几十倍。如果不预热，第一个请求的延迟会非常高，而且显存分配模式也不稳定。预热之后，显存分配器已经建立好了缓存，后续请求的显存分配会快很多，也更稳定。

4.3 并发控制模块的实现

这是核心部分，我把信号量、线程池、显存监控整合在一起：

# app/concurrency.py import asyncio from concurrent.futures import ThreadPoolExecutor import torch class GPUConcurrencyController: def __init__(self, max_concurrent=2, max_workers=2): self.semaphore = asyncio.Semaphore(max_concurrent) self.executor = ThreadPoolExecutor(max_workers=max_workers) self.max_concurrent = max_concurrent async def run(self, func, *args, **kwargs): async with self.semaphore: loop = asyncio.get_event_loop() return await loop.run_in_executor( self.executor, lambda: func(*args, **kwargs) ) def get_gpu_memory_info(self): if not torch.cuda.is_available(): return {} return { "allocated": torch.cuda.memory_allocated() / 1024**3, "reserved": torch.cuda.memory_reserved() / 1024**3, "max_allocated": torch.cuda.max_memory_allocated() / 1024**3, }

这个控制器把信号量和线程池绑在一起，调用的时候只需要await controller.run(inference_func, data)，它会自动处理限流和线程调度。

4.4 主服务的组装

# app/main.py from fastapi import FastAPI, HTTPException from app.model import InferenceModel from app.concurrency import GPUConcurrencyController import torch app = FastAPI() model = InferenceModel() controller = GPUConcurrencyController(max_concurrent=2, max_workers=2) @app.on_event("startup") async def startup(): model.load() @app.post("/predict") async def predict(data: dict): try: input_tensor = preprocess(data) output = await controller.run(model.predict, input_tensor) return {"result": postprocess(output)} except torch.cuda.OutOfMemoryError: raise HTTPException(status_code=503, detail="GPU memory exhausted, please retry later") except Exception as e: raise HTTPException(status_code=500, detail=str(e)) @app.get("/health") async def health(): return { "status": "ok", "gpu_memory": controller.get_gpu_memory_info() }

这里加了两个东西：一个是OOM的异常捕获，返回503让客户端重试；另一个是健康检查接口，可以实时看到显存占用情况，方便排查问题。

4.5 参数调优的实操过程

参数调优不能靠猜，得用数据说话。我的做法是写一个简单的压测脚本，逐步增加并发数，观察显存和延迟的变化：

# 压测脚本示例 import asyncio import aiohttp import time async def send_request(session, url, data): start = time.time() async with session.post(url, json=data) as resp: result = await resp.json() return time.time() - start async def benchmark(concurrency): url = "http://localhost:8000/predict" data = {"input": "test"} async with aiohttp.ClientSession() as session: tasks = [send_request(session, url, data) for _ in range(concurrency)] latencies = await asyncio.gather(*tasks) return latencies # 逐步增加并发 for c in [1, 2, 3, 4, 5]: latencies = asyncio.run(benchmark(c)) print(f"Concurrency {c}: avg={sum(latencies)/len(latencies):.3f}s, max={max(latencies):.3f}s")

跑这个脚本的时候，同时用nvidia-smi -l 1监控显存变化。你会看到随着并发数增加，显存占用逐步上升，延迟也会上升。当显存接近上限或者延迟急剧增加的时候，就说明到了瓶颈，当前的并发数就是上限。

我实测下来，对于ResNet50这种级别的模型，24G卡上并发数设3到4比较合适，再高就会开始出现OOM或者延迟抖动。但你的模型可能不一样，一定要自己测。

5. 常见问题与排查技巧实录

5.1 OOM了但显存看起来还有空余

这是最让人头疼的情况：nvidia-smi显示还有好几个G的空闲显存，但程序就是报OOM。原因通常是显存碎片。PyTorch的显存分配器会缓存已经分配的显存块，如果请求的显存大小和缓存块不匹配，即使总空闲显存够，也分配不出来。

解决办法有几个：一是设置PYTORCH_CUDA_ALLOC_CONF=expandable_segments:True，让分配器支持可扩展的显存段，减少碎片；二是定期调用torch.cuda.empty_cache()，但这会释放缓存，导致后续分配变慢，不能频繁用；三是尽量让每个请求的显存分配模式一致，比如固定输入尺寸，避免动态shape导致的碎片。

5.2 服务跑一段时间后越来越慢

这个问题的原因可能是显存泄漏。虽然PyTorch的Tensor在Python GC的时候会自动释放显存，但如果你的代码里有循环引用，或者把Tensor存到了全局变量里，GC就回收不了。

排查方法是定期打印torch.cuda.memory_allocated()，看它是不是在持续增长。如果是，就用gc.collect()手动触发垃圾回收，看能不能降下来。如果降不下来，那就是有地方持有Tensor的引用没释放。

另外一个可能的原因是CUDA上下文切换开销。如果你的服务同时跑了多个模型，或者频繁地在CPU和GPU之间拷贝数据，上下文切换的开销会累积。解决办法是尽量把数据预处理放在GPU上做，减少拷贝次数。

5.3 并发数设了但实际不生效

有时候你设了Semaphore为3，但实际观察到的并发数只有1或者2。这种情况通常是线程池的max_workers设小了，或者事件循环被阻塞了。

检查方法很简单：在推理函数里打印当前线程ID和时间戳，看是不是真的有多个线程在同时执行。如果没有，那就是线程池的问题。如果事件循环被阻塞了，其他请求连Semaphore都进不来，表现就是请求排队时间很长但GPU利用率不高。

5.4 常见问题速查表

5.5 几个我踩过的坑

第一个坑是在async函数里直接调用同步的CUDA操作。这个会导致事件循环阻塞，表现是服务看起来在跑，但所有请求都卡住。解决办法就是用run_in_executor或者to_thread包一层。

第二个坑是Semaphore的release没放在finally里。有一次推理过程中抛了异常，Semaphore没释放，后续所有请求都卡在acquire上，服务直接假死。后来改成async with就没这个问题了。

第三个坑是模型加载了两次。因为用了@app.on_event("startup")，但在某些部署方式下这个事件会触发多次，导致模型被加载多次，显存直接翻倍。解决办法是在加载前检查模型是否已经加载，或者用单例模式。

第四个坑是没有设置CUDA_VISIBLE_DEVICES。如果机器上有多个GPU，不设置这个环境变量的话，PyTorch默认会用第0张卡，可能跟其他服务冲突。建议在启动脚本里显式指定。

提示：如果你的服务部署在容器里，记得给容器设置--gpus all或者--gpus device=0，否则容器里看不到GPU。另外，容器里的显存限制和宿主机可能不一致，需要额外注意。

6. 进阶方向：从限流到动态批处理

信号量限流能解决OOM问题，但它的吞吐量上限是固定的。如果你的请求量很大，而且对吞吐量有要求，那就需要考虑动态批处理了。

动态批处理的核心思想是：把短时间内到达的多个请求攒成一个batch，一起送给GPU推理，这样GPU的计算单元利用率会高很多。比如单请求推理需要80毫秒，batch size为4的时候可能只需要120毫秒，平均每个请求30毫秒，吞吐量提升了将近3倍。

实现动态批处理需要几个组件：一个请求队列，一个批处理调度器，一个结果分发机制。请求进来之后先入队，调度器根据当前队列长度和等待时间决定是否组batch，推理完成后把结果按请求ID分发回去。这块复杂度不低，我后面会单独写一篇来讲。

另外一个方向是多模型共享GPU。如果你的服务需要同时跑多个模型，可以用显存池化的方式，让多个模型共享同一块显存，按需分配。这个需要更底层的CUDA编程或者用NVIDIA的MPS（Multi-Process Service），门槛比较高，一般场景用不上。

我个人在实际操作中的体会是：先把限流做稳，再考虑优化吞吐。很多团队一上来就想搞动态批处理，结果限流都没做好，服务天天OOM，反而耽误了上线时间。限流是保底，批处理是锦上添花，顺序不能反。