企业官网建设流程全解析

先说结论：把一台GPU推理服务部署出来很简单，但让它扛住几十上百个并发请求不把显存打爆，才是真正考验工程经验的地方。这个坑我踩过不止一次：本地测试一切正常，模型加载完毕、单次推理耗时只有几十毫秒，看起来完美；一旦接上真实流量，请求一多，NVIDIA驱动直接抛一个CUDA OOM，整个服务崩掉，后续请求全部堆积超时。这篇文章就围绕 FastAPI 部署 GPU 推理服务时最核心的并发控制问题展开，讲清楚显存溢出为什么发生、如何设计并发控制策略、以及可以照抄的工程实现。

适合正在做 AI 服务化部署、模型推理接口封装、或者从简单 Demo 过渡到生产环境的读者。如果你也遇到过类似“请求一多就显存溢出”的问题，下面这些分析、代码和排查思路，大概率能帮你省掉几周的摸索时间。

1. 溢出根源：FastAPI 并发模型与 GPU 显存分配的冲突

1.1 FastAPI 天生“并发友好”，但 GPU 推理不吃这一套

FastAPI 基于 asyncio，默认情况下是单进程事件循环。它的并发模型是：I/O 等待期间切换协程，让同一个进程同时处理大量请求。这种设计对数据库查询、外部 HTTP 调用这类“I/O 密集”任务非常合适——CPU 在等待期间可以做别的。但 GPU 推理不是纯 I/O 任务，它分两个阶段：

• 前处理/后处理阶段：例如图像 reszie、归一化、NMS 去重，这些跑在 CPU 上，确实是普通计算。
• 模型计算阶段：张量从 CPU 拷贝到 GPU、显存分配、kernel launch、GPU 执行、结果回拷，这一整个链路会占用显存、占用 GPU 的 SM。

问题出在哪呢？FastAPI 的异步模型只控制“函数何时被调度”，它并不会阻止多个请求同时进入模型推理代码。比如你的推理函数是同步的，FastAPI 的线程池会自动用多个线程去并行执行——如果同时有 10 个请求进来，那就可能有 10 个线程同时往 GPU 上送数据、同时分配显存。

我见过不少人的做法是直接用async def predict()里调用模型推理，这更危险。因为同步推理会阻塞事件循环，FastAPI 会退化为“一个请求跑完才跑下一个”，吞吐量低倒是小事，更麻烦的是阻塞期间所有其他请求全部卡住。而用def同步接口走线程池，GIL 对 PyTorch 的 CUDA 运算不那么致命，因为 CUDA 底层计算会释放 GIL，但显存分配竞争就来了。

用一句直观的话理解：FastAPI 以为自己在“并发”，但 GPU 说“你一次派这么多活，我内存（显存）不够了”。

1.2 显存到底是怎么被“挤爆”的

在深入并发控制之前，必须先搞清楚显存溢出的技术本质。很多人以为显存溢出是“模型太大 + 请求太大”，其实不完全对。运行时显存占用由几部分组成：

• CUDA context 常驻开销：PyTorch 加载后无论模型多大，初始化 CUDA 上下文就会占 300M~600M 显存，这是固定成本。
• 模型权重：每个参数按 FP32 算 4 字节，一个 7B 参数模型光权重就约 28GB，如果加载 FP16 版则减半为 14GB。
• 输入输出与中间激活张量：推理时每一层算出来的中间结果，反向传播虽然不需要，但推理的前向中间张量也会临时占用显存，尤其是长序列文本生成或大分辨率图像，这块往往比模型权重还大。
• CUDA caching allocator 的缓存：PyTorch 默认不会把释放的内存立即还给驱动，而是缓存起来重复利用。这个机制在并发场景下会产生大量显存碎片，峰值占用可能远超实际活跃张量之和。

多请求并发时，每个请求都有一份独立的“中间张量集合”和推理计算图。比如单个请求峰值显存占用 2GB，10 个请求同时推理，理论上就有 20GB 的瞬时占用——不是“排队慢慢算”，而是“同时全在算”。GPU 显存不像系统内存那样可以从磁盘换页，它没有 swap 机制，一旦超出物理容量就是CUDA out of memory，且显存分配失败往往很难恢复，进程进入不稳定状态，即使后续请求减少也可能继续报错。

理解了这一点，你就会明白：要避免显存溢出，核心不是“让模型更小”，而是限制同时进入 GPU 计算的请求数量，让峰值占用可控。

2. 方案选型：五种并发控制思路的取舍

先说结论再说原理。我在实践里把并发控制策略分为五种，按工程代价从低到高排列：

2.1 方案一：全部串行化（最简单，但吞吐太低）

简单粗暴地用一个全局锁（threading.Lock）包住推理函数，所有请求排队进入 GPU。优点极其明显：显存占用恒等于单个请求峰值，绝不溢出，代码只有两行。缺点也极其明显：GPU 利用率极低。

深度学习模型推理有个特点：TensorRT、PyTorch 的 kernel 在 GPU 上执行时，单个 batch 内 SM 不一定满载，尤其是小 batch 和小模型，GPU 绝大多数时间处于“算一下、等数据、再算一下”的状态。串行化相当于把 GPU 并发吞吐能力直接归零，QPS 完全取决于单次推理延迟。

实测比较：一个 YOLOv8 模型单张图推理 20ms，串行化能撑到 50 QPS 就不错了，而 GPU 实际能力可能支持几百 QPS。串行化更适合“显存极其紧张、单次推理很重、又不在乎吞吐”的场景，例如超大模型单实例部署。

2.2 方案二：信号量限制并发上限（最通用）

信号量是并发控制里的经典原语。思路是：允许 N 个请求同时进入推理计算，N 可以根据显存余量计算得出，多余的请求在信号量外等待。

实现方式分为线程信号量和协程信号量两种。如果你的推理函数是同步的，FastAPI 会把它丢给线程池执行，此时要用threading.BoundedSemaphore；如果你的推理函数是异步的，则用asyncio.Semaphore，通过async with包装。

这个方案的优点是显存峰值确定性很强：只要 N 设置正确，峰值占用大约是单请求峰值的 N 倍，加少量缓存波动。GPU 吞吐也能打满到一定程度。缺点是：固定并发数仍然可能让 GPU 利用率不高，尤其模型本身有能力吃更大的 batch，N 个单请求推理反而浪费了计算能力。

2.3 方案三：请求队列 + 批量聚合（吞吐最高的主流方案）

这是目前生产环境最推荐的方案：设置一个后台线程，从队列里收集积压的请求，凑够固定 batch 大小（或等待最大积压时间）再统一送入 GPU 推理。这样显存峰值取决于 batch 大小而非请求总数，同时 GPU 计算效率最高——TensorRT 和 PyTorch 对大 batch 的矩阵运算都有显著优化。

代价是工程复杂度。你需要自己实现队列积压逻辑、超时机制、批量张量拼接和结果分拆。对于使用不同模型架构的场景，batch 拼接代码各不相同。但收益值得，我做过一个 YOLOv5 检测服务，单请求批处理从 12ms 降到 27ms/8 张图（等效单张 3.4ms），吞吐提升约 3.5 倍。

2.4 方案四：多进程隔离 + 显存预算分配

如果服务里同时部署多个模型、或者要隔离不同模型之间的显存竞争，可以考虑多进程部署。每个进程加载独立的模型副本，有自己的显存分配，进程之间互不干扰。配合nvidia-smi的“显存上限”功能（如--memory-limit），可以硬性为每个进程分配显存预算。

这个方案的缺点是模型多副本会成倍增加显存占用，一个 7B 模型你已经 14GB 了，开两个进程就要 28GB。多进程通信（如结果返回）还需要额外的任务队列，部署成本高。我会把它视为“万不得已的隔离方案”，而不是并发控制的首选。

2.5 方案五：外部推理引擎接管调度

如果不想自己实现并发控制和批处理，可以直接使用专业推理引擎，例如 Triton Inference Server、TensorRT、TorchServe 等。这些引擎自带动态批处理器（Dynamic Batching）和并发调度器，把 GPU 利用率优化到了极致。FastAPI 只负责接收 HTTP 请求，把推理任务转发给这些引擎。

实际用下来，Triton 的 Dynamic Batching 效果很好，但对于小团队、单模型场景，额外引入一个 C++/Docker 服务带来的运维负担可能超过它带来的收益。FastAPI 内部实现一个小型动态批处理，其实没有想象中复杂，后面给出可复用代码时要重点理解它的积压窗口机制。

3. 核心实现：从信号量到动态批处理的完整代码

3.1 服务端基础结构：模型预加载与单例管理

实现并发控制前，先搭建服务的基本骨架。核心原则是：

• 模型在 FastAPI 启动时加载一次（lifespan 里完成），避免每次请求都加载模型。
• 模型推理函数绑定到单个实例上，尽量不要在接口函数里临时构造模型对象。

from contextlib import asynccontextmanager import torch import torchvision.models as models from fastapi import FastAPI # 全局单例模型 model = None @asynccontextmanager async def lifespan(app: FastAPI): print("Loading model...") global model model = models.resnet50(pretrained=True).eval().cuda() # 预热 + 断言显存安全 with torch.no_grad(): dummy = torch.randn(1, 3, 224, 224).cuda() model(dummy) torch.cuda.synchronize() print(f"Model loaded, allocated memory: {torch.cuda.memory_allocated() / 1024**2:.1f} MB") yield print("Shutting down...") app = FastAPI(lifespan=lifespan)

预热的细节容易被忽视但我强烈建议保留。CUDA 上下文在第一次推理时才真正初始化，很多 kernel 是惰性编译的，第一次调用比后续慢很多。如果不预热，第一个请求会因为 CUDA 初始化耗时导致超时，而且显存占用变化也发生在第一个请求期间，给并发控制带来不可预期的波动。

3.2 用信号量设置并发上限

先给出信号量方案，它简单可靠，适合快速解决问题。

import asyncio import threading from fastapi import FastAPI from pydantic import BaseModel # 协程信号量：用于限制同时进入 GPU 推理的协程数量 inference_semaphore = asyncio.Semaphore(2) # 最多两个并发推理任务 async def run_inference_with_semaphore(input_tensor): async with inference_semaphore: # 并发请求在此排队 # 同步推理函数放进线程池，避免阻塞事件循环 loop = asyncio.get_running_loop() result = await loop.run_in_executor( None, sync_predict, input_tensor ) return result

注意一个关键点：asyncio.Semaphore(2)限制的是协程数量，而不是线程数量。一个协程占住信号量后，如果内部的同步推理阻塞了，其他协程依然可以继续各自占住信号量。由于推理函数在run_in_executor里执行，GIL 的问题相对小，但显存竞争是实打实的。所以信号量值的设定，永远要小于或等于模型可承受的并发推理数量。

那么并发上限 N 怎么定？一个可靠的估算是：

N = floor((总显存 - 模型权重显存 - CUDA context 预留) / 单请求峰值推理显存增量)

例如一张 24GB 的 3090/4090，模型权重占 6GB，预留 CUDA context 和缓存 2GB，单个请求峰值增加约 2GB，那么 N 取 6~8 是安全的。更保守一点取 4，给诸如数据加载、临时张量留缓冲。

我在实际部署时还会写一个动态检测保护：推理开始前检查torch.cuda.memory_allocated()是否超过阈值，如果超过就人为等待重试。这种“软保护”比裸报 OOM 友好太多。

3.3 动态批处理：队列积压 + 定时聚合

切换到正式生产方案。这个实现把“并发”从请求级别迁移到 batch 级别，核心思路是：

1. 请求进入一个线程安全队列。
2. 后台一个线程每间隔max_wait时间检查队列，积累一批输入。
3. 若队列长度达到batch_size，立即取走这批，拼接为一个批次张量推理。
4. 结果按原始顺序返回给各个等待方。

下面给出一个完整的实现，你可以直接抄过去改模型调用部分：

import queue import threading import time import torch import torch.nn.functional as F from fastapi import FastAPI from pydantic import BaseModel class BatchedInferenceServer: def __init__(self, model, batch_size=8, max_wait=0.01): self.model = model self.batch_size = batch_size self.max_wait = max_wait self.in_queue = queue.Queue() self.result_map = {} # 用 id 关联输入与结果 self.stop = False self.worker = threading.Thread(target=self._loop, daemon=True) def start(self): self.worker.start() def stop(self): self.stop = True def _loop(self): """后台循环：攒一批请求，推理一次""" while not self.stop: batch = [] # 先查一个元素，等 max_wait 攒批 item = self.in_queue.get() batch.append(item) deadline = time.time() + self.max_wait while len(batch) < self.batch_size: remaining = deadline - time.time() if remaining <= 0: break try: # 限时等待后续请求 item = self.in_queue.get(timeout=remaining) batch.append(item) except queue.Empty: break inputs = torch.cat([x["tensor"] for x in batch], dim=0) inputs = inputs.cuda() with torch.no_grad(): outputs = self.model(inputs) outputs_cpu = outputs.cpu() # 按位置把结果放回 for i, req in enumerate(batch): req["result"] = outputs_cpu[i] req["event"].set() def predict(self, input_tensor): """外部调用接口：输入 [C,H,W]，返回 [C] 或分类结果""" input_tensor = input_tensor.unsqueeze(0) # 加 batch 维 req = {"tensor": input_tensor, "event": threading.Event(), "result": None} self.in_queue.put(req) req["event"].wait() # 阻塞等待后台线程完成 return req["result"]

然后 FastAPI 接口这样写：

class InferenceRequest(BaseModel): data: list # 假设是 224*224*3 的数组 @app.post("/predict_batch") async def predict_batch(req: InferenceRequest): tensor = torch.tensor(req.data, dtype=torch.float32).view(1, 3, 224, 224) result = batched_server.predict(tensor) # 这个函数是阻塞的 # 使用 run_in_executor 把它放到线程池， # 否则同步阻塞会卡住事件循环 loop = asyncio.get_running_loop() result = await loop.run_in_executor(None, batched_server.predict, tensor) return {"prediction": result.tolist()}

这里有三个技术细节值得展开：

第一，为什么用threading.Event而不是Future？因为后台 worker 是普通线程，asyncio.Future是协程绑定对象，跨线程传递要手动处理同步。Event轻量且线程安全，无用依赖最少。

第二，max_wait怎么取值？这个值直接决定了“批处理延迟”和“吞吐收益”的平衡。max_wait越短，请求等待时间越短，但 batch 更小；max_wait太长，吞吐高但单请求延迟肉眼可见地增加。我的经验是从0.01秒（10ms）起步，观察 Grafana 上的平均延迟和 GPU 利用率再调。如果模型单次推理是 20ms，max_wait=10ms会让请求在队列里平均多等 5ms，换来 batch 提升 2~4 倍，收益显著。

第三，为什么输出要搬到 CPU 后再拆？GPU 显存是稀缺资源，推理完第一时间把输出拷回 CPU（.cpu()），然后显存里的中间张量可以被 allocator 回收复用。不要在 GPU 上长期保存结果，否则显存占用会随结果集大小线性增长——这是很多人遗漏的隐性泄漏。

这个动态批处理方案结合前面信号量的思路还可以做“并发上限 + 批处理”的双层控制：信号量限制最终进入 GPU 的批次数量（例如最多同时 2 个批次），后台 worker 负责决定每个批次怎么组。生产环境我用这种组合压测，显存曲线非常平稳。

3.4 什么时候应该用多进程？

前文提过多进程隔离，这里给一个具体的适用判断标准。我认为满足以下任一条件时才认真考虑多进程：

• 同时部署两个不同模型，其中一个模型偶发峰值显存会挤压另一个模型的核心 Service。
• 单个模型推理有严重的碎片化问题，进程重启恢复比内存管理更现实。
• 有明确的可用性要求，希望一个进程崩溃不影响整体服务。

多进程方案里，FastAPI 主进程只做 HTTP 解析和任务分发，推理放在子进程。常见做法是 Pythonmultiprocessing+ Queue，或者直接上 Redis/RabbitMQ 排队，推理 worker 作为独立消费者。代价是每个 worker 都有一份 CUDA context 和模型副本，显存倍增。以一个 7B 模型 FP16 为例，一个 worker 占 14GB，两个 worker 就 28GB，没有 40GB+ 的卡根本玩不转。

如果你确定要走多进程，我的建议不是自己造轮子，而是去用现成的torch.multiprocessing写法，或者干脆换 Triton。自己写进程池还要处理 Python 序列化问题，坑非常多。

4. 工程化落地：显存监控、压力测试与参数选择

4.1 别靠猜：实时监控显存和 GPU 利用率

没有监控的并发控制都是盲调。部署 GPU 推理服务前，至少要建立两个监控指标：

• 显存占用（memory_allocated和memory_reserved）
• GPU 利用率（torch.cuda.utilization()或 nvidia-smi 的值）

服务内可以用一段轻量代码周期性记录：

def get_gpu_stats(): allocated = torch.cuda.memory_allocated() / 1024**2 reserved = torch.cuda.memory_reserved() / 1024**2 utilization = torch.cuda.utilization() return {"allocated_mb": round(allocated, 1), "reserved_mb": round(reserved, 1), "gpu_util": utilization}

把它挂到 FastAPI 的/health接口里，配合 Prometheus 抓取。不要用nvidia-smi单独去抓，因为nvidia-smi显示的进程内存是你进程的 reserved（分配到的显存），不是你的 PyTorch allocator 真正在用的 active 内存，两者差距可能很大——reserved 大但 allocated 小，说明缓存碎片严重。

排查显存问题的另一个利器是torch.cuda.memory_summary()。在测试环境里刻意构造并发请求，然后调用这个函数看分配明细，能直观看到cached和active的差异、每个张量在哪一层创建。

4.2 压测方法：明确你要测什么

简单场景用wrk、locust、apache bench都可以压 HTTP 层，但 GPU 服务的压测要特别注意一个变量：并发数不等于 QPS，而我们要控制的是并发数——即“同时有多少请求正在跑推理”。

用locust设置用户数为 10、20、50、100，RPS 峰值会随并发数上升。然后观察两个指标：

• 显存峰值：只要不超过“安全阈值”（通常设为总显存 80%）即可。
• P95 延迟：批处理模式下，P95 会随并发略微上升，这是积压等待导致的正常现象。

一个我常用的压测小技巧：先单独测单请求的显存峰值占用（torch.cuda.max_memory_allocated()），再按前文公式估算安全并发 N，再用 N 去压测，而不是随机选一个并发数。这样可以一次性设置合理的压测目标。下面给出一个压测前后对比的参考表（基于一枚 12GB 显存的 GPU，单请求峰值约 1.5GB，模型占用 4GB）：

注意动态批处理的 P95 延迟高于信号量，因为它要等待攒批，这是延迟换吞吐的取舍。如果你的业务对延迟更敏感（比如实时视频流），可能选信号量更合适；如果业务有批处理天然属性（比如批量图片审核、批量文本扫描），动态批处理优势明显。

4.3 FastAPI 部署参数与 Uvicorn 配置

目标环境用 Uvicorn 部署时，--workers数量很关键。对于 GPU 推理服务，我强烈建议 workers 数为 1，除非你有多个 GPU 或走了多进程推理 worker 方案。

原因很直接：Uvicorn 默认每个 worker 是独立的 Python 进程，各自加载模型并申请显存。你配置--workers 4，等于 4 个进程同时把模型加载进显存，显存线性翻四倍。很多人部署时图省事“并发不够就加 worker”，结果加完直接 OOM 崩溃。

正确做法是：1 个 Uvicorn 进程 + FastAPI 内部线程池（run_in_executor）跑推理 + 信号量/批处理控制显存。如果你需要提升吞吐，优先优化批处理大小和信号量，而不是急着加进程。

另外一个真实踩过的坑是 Uvicorn 的日志丢失。高并发场景下，access_log可能刷得飞快，但某些 worker 的日志神秘消失。排查后发现是 Uvicorn 默认日志 handler 的队列溢出导致的。调试推理服务时日志太重要，建议关掉 Uvicorn 默认 access log（--no-access-log），改用 Python logging 模块输出到带RotatingFileHandler的文件，避免日志挤掉错误输出。

5. 常见问题与避坑实录

5.1 为什么我设了并发数还是 OOM？

最常见的原因是“你以为并发是 N，实际是 N+1”。常见来源有两个：

• FastAPI 接口如果写成async def，内部又直接调同步推理，那么并发上限完全不可控，因为协程切换点不可预测。正确做法是接口只负责解析和组装，推理统一走信号量或队列。
• run_in_executor默认线程池大小是 32，如果你在信号量之外调用了run_in_executor，且信号量没生效，那就相当于同时创建了 32 个待执行推理任务，虽然它们可能排队，但模型已加载后所有线程都在尝试分配显存上下文。

我的经验是：把信号量放在最靠近“实际进行model(input)调用”的地方，而不是放在接口层。宁可多包一层函数，也不要只在接口层放锁。

5.2 显存一直上涨不下降，是泄漏吗？

不一定是泄漏。PyTorch 的 caching allocator 会慢慢扩大缓存（memory_reserved），当你请求量下降后，它不会立刻把显存还给 GPU。这一点会让很多人误判为“内存泄漏”。区分方法很简单：跑完高并发压测后，持续发少量请求，观察显存占用。如果显存稳定在某个值不变，而不是持续上涨到 OOM，那就不是泄漏，是正常的 allocator 缓存行为。

如果确认有泄漏，重点检查以下三处：

• 推理循环里是否创建了新的 CUDA 图或计算图（torch.no_grad()是否包裹完整推理过程）。
• NMS、后处理等操作是否有张量retain_grad=True，导致中间变量保留。
• 是否用torch.cuda.memory_summary()查看是否有大量allocated但未释放的 Tensor。

5.3 PYTHON 的 GIL 会把多线程推理拖垮吗？

结论先行：对大多数 PyTorch GPU 推理任务，GIL 不会成为瓶颈，因为 CUDA kernel 运行时会释放 GIL，线程主要的时间在等待 GPU 计算结果。真正受 GIL 影响的可能是前处理阶段——如果你的前处理大量使用纯 Python 循环进行 numpy 操作，那这部分无法并行。

优化建议：前处理尽量向量化（用 numpy/PyTorch 张量运算），不要用 Python 逐像素循环。同时把前处理做在接口线程里、推理放进线程池，这样 CPU 密集的前处理和 GPU 推理可以重叠执行。实测一个检测服务，前处理从 8ms 优化到 1.2ms，整体吞吐提升明显。

5.4 OOM 崩溃后服务怎样才能自动恢复？

GPU 服务 OOM 后进入不稳定状态，最简单的可靠恢复手段是：进程退出，重新拉起。所以生产环境中要注意两点：

• 不要把显存用“满格”，默认安全水位设为总显存 80%。留出 20% 的余量给紧急情况下的内存峰值。
• 给服务加一个健康检查（/health），检查内容包含“最近一次推理是否成功”的状态位。一旦连续失败 N 次，返回 503，让 orchestrator（K8s/Supervisor）自动重启进程。

我自己的实践是写了一个轻量 watchdog：每 30 秒做一次小规模预热推理，如果出现 CUDA OOM，进程主动os._exit(1)，靠外部 supervisor 拉起。这个方法比在异常里 try-except 后强制 continue 可靠得多——因为 CUDA 上下文已经损坏，勉强继续反而会因为状态错乱产生更诡异的问题。

6. 现有生态：几种辅助方案的实际体验

6.1 用 Ollama 之类的本地推理引擎兜底

如果你不想手写推理代码，一个很务实的选择是用 Ollama 或 LocalAI 这类本地推理引擎，FastAPI 只需要调用它们暴露的 HTTP 接口。这么做的好处是引擎内部已经做了并发队列管理，Ollama 默认会串行化请求（或其他策略），天然避免了多请求同时抢显存的问题。

FastAPI 调 Ollama 的代码非常少：

import httpx from fastapi import FastAPI app = FastAPI() OLLAMA_URL = "http://localhost:11434/api/generate" @app.post("/generate") async def generate(prompt: str): async with httpx.AsyncClient(timeout=120) as client: resp = await client.post(OLLAMA_URL, json={"model": "llama3", "prompt": prompt, "stream": False}) return resp.json()

这个方案的缺点是你丢失了批处理控制的灵活性，无法根据自己的显存余量精细控制 batch 大小。适合“不想动太多底层、模型固定、请求量不大”的团队。

6.2 Flask 与 FastAPI：并发模型差异

很多项目从 Flask 迁到 FastAPI 就是因为 Flask 的异步支持不是原生能力。但我要泼一盆冷水：如果推理代码本身是同步的，FastAPI 和 Flask + gunicorn 多 worker 的差异并没有想象中那么大。Flask 默认多线程模型下，并发请求直接用线程池处理，没有事件循环的复杂性，你用threading.Semaphore信号量也能实现类似的显存控制。

FastAPI 的优势主要在于：原生异步接口带来更细粒度的协程控制、Pydantic 请求校验、以及 OpenAPI 文档自动生成。但请注意：如果你把推理函数丢进run_in_executor，你实际上写的还是“线程 + 信号量”模型，和 Flask 的多线程差异不大。所以不要为了“FastAPI 是异步的”这句话而忽略底层的并发控制设计。

6.3 面向 GPU 调度的后续扩展

并发控制做稳以后，你还可以考虑 GPU 调度层面的事情。比如多个服务共享同一张卡时，可以用CUDA_VISIBLE_DEVICES指定进程可见的 GPU 编号，或者用 MIG（多实例 GPU）把物理 GPU 切分成多个实例。这些对显存隔离很有帮助，但前提是你已经做好单进程内的并发控制，否则实例再多也会被打满。

个人收官体会

做了多次 GPU 推理服务部署后，我现在看待“显存溢出”的态度变成了：它不是一个需要反复 kill 的 Bug，而是一个需要提前设计的容量规划问题。每次面对一个新模型，我做的第一件事是在 Jupyter 或脚本里测三件事——单请求峰值显存、模型权重显存、预热后的基线显存，然后基于这三样去做并发数的数学计算，而不是贸然上线再让监控告警推着我救火。

这个思路相信对你也适用：先摸清你的显存“家底”，再用信号量和动态批处理建好“闸门”，最后让监控和压测告诉你剩余吞吐空间在哪里。把并发控制做好之后，你会发现 GPU 推理服务不仅稳定了，吞吐还可能比你原来无脑并发时更高——因为批处理带来的收益，远超你的直觉。