企业官网建设流程全解析

1. 从一次线上事故说起：为什么并发控制是GPU推理服务的生死线

去年冬天，我帮一个做图像识别的团队排查线上问题。他们的服务用FastAPI包了一个YOLO模型，部署在一台单卡A10上，平时跑得好好的，结果某天市场部做活动，流量瞬间翻了五倍，服务直接挂了。日志里赫然写着CUDA out of memory，显存被瞬间打满，连带着把同卡上的其他服务也拖死了。事后复盘，问题根本不在模型本身，而在于他们压根没做并发控制——每个请求进来就无脑往GPU上塞，显存不炸才怪。

这个场景太典型了。FastAPI本身是异步框架，天生适合做高并发接口，但GPU是个独占型资源，它的显存容量是固定的，计算单元也是有限的。你不可能像CPU那样开几十个线程同时跑，GPU的并发模型和CPU完全不是一回事。所以，FastAPI + GPU推理这个组合，核心矛盾就在于：Web层的异步高并发和GPU层的串行独占之间，需要一个中间层来做调度和限流。这个中间层做得好不好，直接决定了你的服务是稳定运行还是三天两头崩。

这篇文章就是围绕这个矛盾展开的。我会从架构设计、并发控制策略、显存管理、实操代码、问题排查几个维度，把“FastAPI GPU推理并发控制”这件事讲透。适合谁看？如果你正在用FastAPI部署深度学习模型，或者准备把本地的推理脚本包装成线上服务，又或者你已经上线了但经常遇到显存溢出、请求超时、服务崩溃的问题，那这篇内容就是为你写的。我会尽量用大白话把原理讲清楚，同时给出可以直接抄的代码和配置，让你看完就能动手改自己的项目。

2. 整体架构设计：FastAPI和GPU之间到底该怎么配合

2.1 为什么不能直接在路由函数里跑推理

很多人的第一版代码是这样的：写一个/predict接口，在函数里直接加载模型、处理输入、调用model()、返回结果。本地测试没问题，一上线就出事儿。原因很简单，FastAPI默认用Uvicorn跑，Uvicorn是异步服务器，它的事件循环是单线程的。你在路由函数里写同步的GPU推理代码，整个事件循环就被阻塞了，其他请求全部排队等着。这时候如果并发量上来，请求队列越堆越长，每个请求都占着内存不释放，最后要么超时，要么OOM。

更糟糕的是，如果你用了async def定义路由，但在里面调用了同步的GPU推理函数，FastAPI会把它丢到线程池里跑。线程池默认是40个线程，意味着最多有40个请求同时往GPU上塞。一张24G显存的卡，跑一个YOLOv8-large大概占4G，40个并发就是160G，显存直接爆炸。所以，绝对不能把GPU推理裸奔在路由函数里，必须加一层控制。

2.2 三种主流方案对比：信号量、队列、独立推理进程

我试过三种方案，各有优劣，下面用表格对比一下。

我个人的选择是请求队列 + 单推理协程的方案。理由是这样既能控制并发数（队列消费者只有一个，天然串行），又能通过队列长度做背压，还能在消费者里做批处理优化。信号量方案虽然简单，但它只是限制了同时进入推理的请求数，排队中的请求依然占着内存，而且没法做批处理。独立进程方案太重了，除非你有多个模型要跑，否则没必要。

2.3 核心设计：一个队列、一个消费者、一个显存监控

我的架构是这样的：FastAPI启动时创建一个asyncio.Queue，同时启动一个后台协程作为消费者。每个请求进来后，把输入数据和一个Future对象打包丢进队列，然后await future等待结果。消费者协程从队列里取任务，调用GPU推理，把结果设置到Future里。这样所有GPU调用都在同一个协程里串行执行，显存占用是可控的。

同时，我会加一个显存监控协程，定期用pynvml查显存使用率。如果超过阈值（比如90%），就暂停从队列取任务，等显存降下来再继续。这个机制能有效防止显存泄漏导致的累积溢出。另外，队列本身要设一个最大长度，比如100，满了之后新请求直接返回503，避免无限堆积。

提示：队列消费者协程要用asyncio.create_task启动，并且要处理好异常，不能让消费者挂掉。消费者挂了整个服务就废了。

3. 核心细节解析：显存管理、批处理与超时控制

3.1 显存到底被谁吃了：模型权重、激活值、缓存

要控制显存，先得知道显存花在哪了。以PyTorch为例，显存占用主要分三块：模型权重、前向传播的激活值、CUDA上下文和缓存。模型权重是固定的，加载后就占着，比如ResNet50大概100M，YOLOv8x大概260M。激活值跟batch size和输入尺寸有关，batch越大、输入越大，激活值越多。CUDA上下文大概占300-500M，这是固定开销。

很多人忽略的是PyTorch的缓存分配器。PyTorch为了加速显存分配，会缓存已经释放的显存块，不立即还给系统。所以你用nvidia-smi看显存占用很高，但实际上模型可能只用了很少。这时候如果你用torch.cuda.empty_cache()可以释放缓存，但频繁调用会影响性能。我的建议是：在服务启动时预热一次，让缓存稳定下来，之后不要频繁清缓存。

还有一个坑是多线程下的显存分配。如果你用了线程池，每个线程都可能触发显存分配，导致缓存碎片化。所以一定要确保GPU推理在单线程/单协程里执行。

3.2 批处理：提升吞吐的关键，但别贪心

批处理是提升GPU利用率最有效的手段。单张图片推理可能只要5ms，但GPU利用率只有10%，因为大部分时间在等数据传输。如果把32张图片拼成一个batch，推理时间可能只要20ms，吞吐量提升8倍。但batch不是越大越好，因为显存占用是线性增长的。你需要找到一个平衡点。

我的做法是：在队列消费者里做动态批处理。消费者不是取一个任务就推理一次，而是等一小段时间（比如10ms），把队列里攒的任务凑成一个batch。如果队列里任务多，batch就大；任务少，batch就小。这样既能提升吞吐，又不会因为固定大batch导致显存溢出。具体实现可以用asyncio.wait_for加超时，或者用queue.qsize()判断。

计算最大batch size的公式大概是：max_batch = (可用显存 - 模型权重 - CUDA上下文) / 单样本激活值。单样本激活值需要实测，你可以用不同batch跑一下，看显存增长曲线。一般来说，留20%的显存余量比较安全。

3.3 超时控制：别让一个慢请求拖死整个队列

队列方案有个隐患：如果某个请求推理特别慢（比如输入图片特别大），它会占着消费者，后面的请求全部等着。所以必须加超时。我的做法是给每个任务设一个超时时间，比如5秒。消费者取到任务后，用asyncio.wait_for包住推理调用，超时就抛异常，把Future设置成超时错误。同时，在请求端也要设超时，如果await future超过一定时间（比如10秒），直接返回504，不等了。

另外，队列本身也要设最大等待时间。如果队列里任务的平均等待时间超过阈值，说明系统过载了，应该直接拒绝新请求。这个可以通过监控队列长度和消费者处理速度来动态调整。

注意：超时时间要根据你的模型和业务来定。实时性要求高的场景，超时要短；离线批处理场景，可以长一点。但无论如何，不能无限等。

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

4.1 项目目录结构与环境准备

先看目录结构，我习惯这样组织：

project/ ├── app/ │ ├── __init__.py │ ├── main.py # FastAPI入口 │ ├── config.py # 配置 │ ├── inference.py # 推理核心逻辑 │ ├── queue_manager.py # 队列管理 │ └── monitor.py # 显存监控 ├── models/ │ └── yolov8n.pt # 模型文件 ├── requirements.txt └── run.sh

环境准备很简单，装这几个包就行：

pip install fastapi uvicorn torch torchvision pynvml numpy opencv-python

如果你用YOLO，再加个ultralytics。注意PyTorch要装GPU版，别装成CPU版了。装完之后用torch.cuda.is_available()验证一下。

4.2 核心代码：队列管理器与推理消费者

先写队列管理器。这个类负责创建队列、启动消费者、提交任务。

import asyncio import logging from typing import Any, Optional logger = logging.getLogger(__name__) class InferenceQueue: def __init__(self, max_size: int = 100, num_workers: int = 1): self.queue = asyncio.Queue(maxsize=max_size) self.num_workers = num_workers self.workers = [] self._running = False async def start(self, inference_fn): self._running = True for i in range(self.num_workers): task = asyncio.create_task(self._worker(inference_fn, i)) self.workers.append(task) async def _worker(self, inference_fn, worker_id: int): while self._running: try: item = await self.queue.get() if item is None: break data, future = item try: result = await inference_fn(data) if not future.done(): future.set_result(result) except Exception as e: logger.exception(f"Worker {worker_id} inference error") if not future.done(): future.set_exception(e) finally: self.queue.task_done() except asyncio.CancelledError: break except Exception: logger.exception(f"Worker {worker_id} unexpected error") async def submit(self, data: Any, timeout: float = 10.0) -> Any: loop = asyncio.get_event_loop() future = loop.create_future() try: self.queue.put_nowait((data, future)) except asyncio.QueueFull: raise RuntimeError("Queue is full, please retry later") try: return await asyncio.wait_for(future, timeout=timeout) except asyncio.TimeoutError: raise RuntimeError("Inference timeout") async def stop(self): self._running = False for _ in self.workers: await self.queue.put(None) await asyncio.gather(*self.workers, return_exceptions=True)

这个队列管理器有几个关键点：maxsize控制队列长度，满了直接抛异常；submit方法带超时，防止无限等待；worker里用try/except包住推理，保证一个任务出错不影响后续任务。

4.3 推理函数：加载模型、预处理、批处理

推理函数是核心。我这里用YOLO举例，但思路通用。

import torch import numpy as np from PIL import Image class InferenceEngine: def __init__(self, model_path: str, device: str = "cuda"): self.device = device self.model = self._load_model(model_path) self.model.eval() # 预热 self._warmup() def _load_model(self, path): from ultralytics import YOLO model = YOLO(path) model.to(self.device) return model def _warmup(self): dummy = np.random.randint(0, 255, (640, 640, 3), dtype=np.uint8) for _ in range(3): self.model(dummy, verbose=False) torch.cuda.synchronize() async def infer(self, image: np.ndarray): # 这里用run_in_executor把同步推理放到线程池，避免阻塞事件循环 loop = asyncio.get_event_loop() result = await loop.run_in_executor(None, self._infer_sync, image) return result def _infer_sync(self, image: np.ndarray): with torch.no_grad(): results = self.model(image, verbose=False) return results[0].boxes.data.cpu().numpy()

注意这里用了run_in_executor，因为YOLO的推理是同步的，直接调用会阻塞事件循环。虽然我们只有一个worker，但放到线程池里可以让事件循环继续处理其他请求（比如健康检查）。不过要小心，如果线程池里同时有多个推理任务，还是会并发。所以我的做法是：队列worker只有一个，它串行地调用infer，而infer内部用线程池执行同步推理。这样实际上GPU调用还是串行的，但事件循环不被阻塞。

4.4 显存监控与动态限流

显存监控用pynvml实现：

import pynvml import asyncio class GPUMonitor: def __init__(self, device_id: int = 0, threshold: float = 0.9): pynvml.nvmlInit() self.handle = pynvml.nvmlDeviceGetHandleByIndex(device_id) self.threshold = threshold self._paused = False def get_memory_info(self): info = pynvml.nvmlDeviceGetMemoryInfo(self.handle) return info.used / info.total async def monitor_loop(self, queue: InferenceQueue): while True: usage = self.get_memory_info() if usage > self.threshold and not self._paused: self._paused = True logger.warning(f"GPU memory usage {usage:.2%}, pausing queue") elif usage < self.threshold - 0.1 and self._paused: self._paused = False logger.info(f"GPU memory usage {usage:.2%}, resuming queue") await asyncio.sleep(1)

然后在worker里检查_paused状态，如果暂停了就等一会儿再取任务。这个机制能有效防止显存缓慢增长导致的溢出。

4.5 FastAPI入口：整合所有组件

最后是FastAPI的入口：

from fastapi import FastAPI, UploadFile, HTTPException from contextlib import asynccontextmanager import numpy as np import cv2 queue_manager = None engine = None monitor = None @asynccontextmanager async def lifespan(app: FastAPI): global queue_manager, engine, monitor engine = InferenceEngine("models/yolov8n.pt") queue_manager = InferenceQueue(max_size=100, num_workers=1) await queue_manager.start(engine.infer) monitor = GPUMonitor(threshold=0.9) asyncio.create_task(monitor.monitor_loop(queue_manager)) yield await queue_manager.stop() app = FastAPI(lifespan=lifespan) @app.post("/predict") async def predict(file: UploadFile): contents = await file.read() nparr = np.frombuffer(contents, np.uint8) image = cv2.imdecode(nparr, cv2.IMREAD_COLOR) if image is None: raise HTTPException(status_code=400, detail="Invalid image") try: result = await queue_manager.submit(image, timeout=10.0) except RuntimeError as e: raise HTTPException(status_code=503, detail=str(e)) return {"boxes": result.tolist()}

这样整个服务就搭起来了。启动命令：

uvicorn app.main:app --host 0.0.0.0 --port 8000 --workers 1

注意--workers一定要设成1，因为多个worker会创建多个进程，每个进程都会加载模型，显存直接翻倍。如果你要多worker，得用共享内存或者独立推理服务的方式。

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

5.1 显存溢出排查速查表

5.2 请求超时和队列堆积怎么破

队列堆积通常是因为消费者处理速度跟不上请求速度。这时候你要看两个指标：队列长度和单请求处理时间。如果队列长度持续增长，说明系统过载了。解决方案有几个：一是加机器，多卡并行；二是优化模型，用更小的模型或量化；三是做限流，在入口层用令牌桶或漏桶算法限制QPS。

超时问题往往是某个请求特别慢导致的。你可以在推理函数里加日志，记录每个请求的处理时间。如果发现某些请求特别慢，可能是输入尺寸太大，可以在预处理阶段做resize。

5.3 那些年我踩过的坑

第一个坑是在异步函数里直接调用同步GPU代码。我一开始写了个async def predict，里面直接model(image)，结果事件循环被阻塞，健康检查都超时了。后来改成run_in_executor才解决。

第二个坑是忘了设队列最大长度。有次压测，请求疯狂进来，队列无限增长，内存直接爆了。后来加了maxsize=100，满了就返回503，服务反而稳定了。

第三个坑是多worker导致显存翻倍。我一开始用--workers 4，结果四个进程各加载一个模型，显存直接满了。后来改成单worker，用队列做并发控制，效果更好。

第四个坑是PyTorch缓存导致显存虚高。nvidia-smi显示显存用了20G，但实际上模型只用了5G。后来用torch.cuda.memory_allocated()看实际用量，才发现是缓存。这个不影响运行，但会误导排查。

提示：排查显存问题时，一定要区分reserved和allocated。allocated是实际用的，reserved是缓存。用torch.cuda.memory_summary()可以看详细分布。

5.4 性能优化的几个小技巧

第一个技巧是用半精度。model.half()可以把显存占用减半，推理速度也能提升。但要注意，有些模型对精度敏感，可能会掉点。可以先在测试集上验证一下。

第二个技巧是用TensorRT或ONNX Runtime。这两个推理引擎比原生PyTorch快很多，显存占用也更低。但转换过程有点麻烦，适合对性能要求高的场景。

第三个技巧是输入预处理放到CPU。图片解码、resize这些操作在CPU上做，不要占用GPU。GPU只负责推理。

第四个技巧是用固定尺寸输入。动态尺寸会导致每次推理都重新分配显存，容易产生碎片。固定成640x640或者416x416，显存分配更稳定。

6. 进阶扩展：从单卡到多卡，从单模型到多模型

6.1 多卡并行的两种思路

如果你有多张卡，有两种方案。一种是数据并行，每个卡跑一个模型副本，请求分发到不同卡上。这种适合模型小、并发高的场景。另一种是模型并行，把大模型拆到多张卡上，适合单卡放不下的大模型。数据并行实现简单，用torch.nn.DataParallel或者手动分发就行。模型并行复杂得多，一般用accelerate或deepspeed。

我的建议是：如果单卡能放下模型，优先用数据并行。每张卡一个队列，请求按轮询或最少连接分发。这样线性扩展，实现也简单。

6.2 多模型服务的路由策略

如果你的服务要跑多个模型，比如一个检测模型、一个分类模型，可以用一个队列但不同worker处理不同模型，或者每个模型一个队列。我倾向于每个模型一个队列，因为不同模型的显存占用和推理时间不一样，分开控制更灵活。路由层根据请求的模型名分发到对应队列。

6.3 监控与告警：别等崩了才发现

生产环境一定要加监控。我一般监控这几个指标：GPU利用率、显存使用率、队列长度、请求延迟、错误率。用Prometheus + Grafana做可视化，设置告警阈值。比如显存超过90%持续30秒就告警，队列长度超过50就告警。这样能在问题恶化前介入。

另外，日志要打全。每个请求的ID、输入尺寸、推理时间、显存变化都记下来。出问题的时候，这些日志就是排查的依据。

6.4 容器化部署的注意事项

如果用Docker部署，要注意几点。一是基础镜像要带CUDA，比如nvidia/cuda:12.1-runtime。二是启动时要加--gpus all。三是显存限制不能用--memory，要用NVIDIA_VISIBLE_DEVICES或者CUDA_VISIBLE_DEVICES。四是容器内nvidia-smi可能不可用，要装pynvml来监控。

还有一个坑是容器内的共享内存。PyTorch的DataLoader默认用共享内存，如果/dev/shm太小会报错。启动时加--shm-size=8g。

我个人在实际操作中的体会是，并发控制这件事，核心不是技术多复杂，而是要对GPU的脾气有敬畏心。它不像CPU那样可以随便压榨，你得顺着它的性子来。队列、限流、监控，这三板斧下去，基本能解决90%的显存溢出问题。剩下的10%，靠的是对模型和业务的深入理解，知道什么时候该批处理，什么时候该降级，什么时候该拒绝请求。这些经验，都是在一次次线上事故里攒出来的。