尧图网页设计 
1. 从一次线上事故说起为什么并发控制是GPU推理服务的生死线去年冬天我帮一个团队排查他们AI绘画服务的线上故障。现象很典型单张图片生成耗时约3秒平时跑得好好的一到晚上高峰期服务就像被掐住脖子一样请求排队越来越长最后直接返回500错误。登录服务器一看nvidia-smi显示显存占用100%但GPU利用率只有30%出头——典型的显存溢出导致的进程崩溃重启循环。这个场景在FastAPI GPU推理的架构里太常见了。FastAPI本身是异步框架天生适合处理高并发IO密集型任务但GPU推理是计算密集型而且显存是独占资源。当你用FastAPI直接暴露一个推理接口每个请求都去加载模型或者分配显存并发一上来显存瞬间被撑爆服务直接挂掉。这篇文章就是围绕这个问题展开的。我会从架构设计、并发控制策略、显存管理、实操代码、踩坑经验几个维度把FastAPI GPU推理并发控制这件事讲透。适合正在用FastAPI做AI服务后端、遇到显存溢出问题的开发者也适合准备把本地推理模型部署成API服务的朋友。读完你至少能搞清楚三件事为什么并发会炸显存、怎么用信号量和队列控制并发、以及怎么在FastAPI里优雅地管理GPU资源。2. 问题根源拆解FastAPI的异步模型与GPU推理的冲突2.1 FastAPI的异步机制到底在异步什么FastAPI基于Starlette和Pydantic核心卖点是async def带来的高并发能力。但很多人对“异步”有误解以为异步就是“同时处理很多事”。准确地说异步是“在等待IO的时候去处理其他请求”。比如数据库查询、HTTP调用、文件读写这些操作CPU大部分时间在等异步可以把这些等待时间利用起来。但GPU推理不一样。当你调用model.generate()或者model.predict()CPU把数据传给GPUGPU开始计算然后CPU等待结果返回。这个等待过程中GPU在满负荷运转显存被模型权重和中间激活值占满。如果你用async def定义推理接口FastAPI会认为这是个异步任务多个请求可以同时进来每个请求都去调用GPU显存瞬间爆炸。我见过太多人这样写app.post(/infer) async def infer(request: InferRequest): result model(request.input) # 同步阻塞调用 return {result: result}这段代码的问题在于async def里面跑了同步阻塞的GPU推理。事件循环被阻塞不说更致命的是多个请求会同时进入这个函数每个都去抢GPU资源。模型可能只加载了一份但推理过程中的中间张量、KV Cache、批处理缓冲区都是按请求分配的并发一高显存直接溢出。2.2 显存溢出的三种典型触发路径根据我处理过的案例显存溢出通常走这三条路第一条模型重复加载。有些实现为了“隔离”每个请求都重新加载模型或者创建新的推理会话。一个7B参数的模型FP16精度大约占14GB显存两个请求同时加载就是28GB一张24GB的卡直接跪。第二条批处理无上限。为了提升吞吐把多个请求攒成一批送进GPU。但批大小没有上限控制显存占用随批大小线性增长。我见过一个服务把batch_size设成64结果单次推理就吃了40GB显存。第三条KV Cache累积。大语言模型推理时KV Cache会随着生成长度增长。如果并发请求多每个请求都维护自己的KV Cache显存占用是并发数乘以单请求Cache大小。并发32路、每路2GB Cache就是64GB远超单卡容量。这三条路径的共同点是显存分配发生在请求处理过程中而不是服务启动时。FastAPI的并发模型让多个请求同时进入处理流程每个请求都去申请显存最终超过物理上限。2.3 为什么不能简单加个锁有人会说那我加个threading.Lock不就行了理论上可以但实际效果很差。锁的粒度如果太大所有请求串行执行GPU利用率上不去吞吐量惨不忍睹。锁的粒度如果太小又起不到保护显存的作用。更合理的做法是在服务层面控制同时进入GPU推理的请求数量让并发数匹配显存容量。这就是并发控制的核心思路。具体来说需要解决三个问题怎么限制并发数、怎么排队、怎么在请求间公平调度。3. 并发控制方案选型从信号量到推理队列3.1 方案一asyncio.Semaphore信号量控制最直接的办法是用asyncio.Semaphore限制同时执行的推理任务数量。信号量的值根据显存容量和单次推理显存占用计算得出。假设一张24GB显存的卡模型权重占14GB系统预留2GB剩余8GB可用于推理中间态。单次推理峰值额外占用2GB那么信号量可以设为4。这样最多4个请求同时进入GPU推理显存占用控制在1424×224GB以内。代码实现大概是这样import asyncio from fastapi import FastAPI app FastAPI() GPU_SEMAPHORE asyncio.Semaphore(4) app.post(/infer) async def infer(request: InferRequest): async with GPU_SEMAPHORE: result await run_in_executor(model, request.input) return {result: result}这里有个关键点run_in_executor把同步的GPU推理放到线程池里执行避免阻塞事件循环。信号量保证同时最多4个任务在跑。这个方案的优点是简单、直观、容易实现。缺点是信号量是进程内的如果用了多个worker进程比如uvicorn --workers 4每个进程有自己的信号量总并发数会翻倍。另外信号量不区分请求优先级先到的先执行后到的干等。3.2 方案二请求队列加消费者模型更工程化的做法是引入一个请求队列推理任务作为消费者从队列里取任务执行。FastAPI接口只负责把请求塞进队列并等待结果不直接碰GPU。这种架构的好处是并发数完全由消费者数量控制和FastAPI的worker数量解耦可以实现优先级队列、超时丢弃、批量合并等高级功能显存管理更可控因为只有消费者进程会碰GPU。实现上可以用asyncio.Queue做进程内队列也可以用Redis、RabbitMQ做跨进程队列。对于单机多卡场景我推荐用进程内队列加多消费者每个消费者绑定一张GPU。import asyncio from concurrent.futures import ThreadPoolExecutor class InferenceQueue: def __init__(self, num_workers2): self.queue asyncio.Queue(maxsize100) self.executor ThreadPoolExecutor(max_workersnum_workers) self.workers [] async def start(self): for _ in range(2): self.workers.append(asyncio.create_task(self.worker())) async def worker(self): while True: request, future await self.queue.get() try: loop asyncio.get_event_loop() result await loop.run_in_executor( self.executor, model_infer, request ) future.set_result(result) except Exception as e: future.set_exception(e) finally: self.queue.task_done()这个方案比信号量复杂但扩展性更好。队列长度可以控制背压消费者数量可以精确匹配显存容量还能做优雅关闭。3.3 方案三动态批处理加显存感知调度如果追求极致吞吐可以考虑动态批处理。核心思路是消费者从队列里取任务时不只取一个而是把当前队列里所有等待的任务攒成一批根据显存余量决定批大小一次性送进GPU。这个方案对显存管理要求最高需要实时监控显存占用动态调整批大小。实现上可以用pynvml读取显存信息结合模型推理的显存profile计算出当前能承受的最大批大小。我实测下来动态批处理在请求量波动大的场景下吞吐提升明显但实现复杂度也高。如果团队没有专门的推理优化经验建议先从信号量或固定消费者队列做起稳定后再考虑动态批处理。3.4 三种方案对比与选型建议方案实现复杂度并发控制精度吞吐量适用场景信号量低中中单进程、请求量稳定请求队列中高中高多worker、需要优先级动态批处理高高高请求波动大、追求吞吐选型建议如果你的服务是单进程部署请求量不大信号量足够用。如果用了多worker或者需要更精细的控制上请求队列。动态批处理适合有专门优化团队、对吞吐有极致要求的场景。4. 实操落地用信号量加线程池实现稳定的GPU推理服务4.1 环境准备与依赖安装先确认基础环境。Python 3.9以上PyTorch或ONNX Runtime按你的模型框架选。FastAPI和Uvicorn是必须的。pip install fastapi uvicorn[standard] pynvml pip install torch torchvision --index-url https://download.pytorch.org/whl/cu118pynvml用来读取显存信息方便计算信号量初始值。如果你用的是NVIDIA显卡驱动版本建议535以上CUDA版本和PyTorch版本要匹配。4.2 显存容量测算与信号量计算这一步是整个方案的基础。你需要搞清楚三个数模型权重显存、单次推理峰值显存、系统预留显存。模型权重显存可以这样估算参数量乘以精度字节数。FP16是2字节FP32是4字节INT8是1字节。比如7B模型FP16就是7×10^9×214GB。单次推理峰值显存包括中间激活值、KV Cache、批处理缓冲区。这个最好用torch.cuda.max_memory_allocated()实测。跑一次推理记录峰值多跑几次取最大值。系统预留显存建议留2-4GB给CUDA上下文、显存碎片、其他进程用。信号量计算公式信号量 floor((总显存 - 模型权重 - 系统预留) / 单次推理峰值)举个例子24GB卡7B模型FP16占14GB系统预留3GB单次推理峰值1.5GB。信号量 floor((24-14-3)/1.5) floor(7/1.5) 4。注意这个计算是保守估计。实际部署时建议先用小信号量跑压测逐步调大观察显存占用和错误率。4.3 完整代码实现与逐行解析下面是一个可直接运行的完整示例。我用了一个模拟的推理函数你替换成自己的模型调用即可。import asyncio import time from concurrent.futures import ThreadPoolExecutor from contextlib import asynccontextmanager import torch from fastapi import FastAPI, HTTPException from pydantic import BaseModel # 全局资源 GPU_SEMAPHORE: asyncio.Semaphore None INFER_EXECUTOR: ThreadPoolExecutor None MODEL None class InferRequest(BaseModel): prompt: str max_tokens: int 128 class InferResponse(BaseModel): result: str latency_ms: float def load_model(): 加载模型实际替换成你的模型加载逻辑 # 这里用sleep模拟加载耗时 time.sleep(2) return {name: mock-model, device: cuda} def model_infer(prompt: str, max_tokens: int) - str: 同步推理函数运行在线程池里 # 模拟GPU推理耗时 time.sleep(0.5) return fgenerated: {prompt[:20]}... asynccontextmanager async def lifespan(app: FastAPI): global GPU_SEMAPHORE, INFER_EXECUTOR, MODEL # 启动时加载模型 MODEL load_model() # 根据显存计算信号量这里硬编码为4 GPU_SEMAPHORE asyncio.Semaphore(4) # 线程池大小等于信号量避免线程堆积 INFER_EXECUTOR ThreadPoolExecutor(max_workers4) yield # 关闭时清理 INFER_EXECUTOR.shutdown(waitTrue) torch.cuda.empty_cache() app FastAPI(lifespanlifespan) app.post(/infer, response_modelInferResponse) async def infer(request: InferRequest): start time.time() try: async with GPU_SEMAPHORE: loop asyncio.get_event_loop() result await loop.run_in_executor( INFER_EXECUTOR, model_infer, request.prompt, request.max_tokens ) except Exception as e: raise HTTPException(status_code500, detailstr(e)) latency (time.time() - start) * 1000 return InferResponse(resultresult, latency_mslatency) app.get(/health) async def health(): return {status: ok, gpu_available: torch.cuda.is_available()}逐行解析几个关键点lifespan是FastAPI推荐的生命周期管理方式替代了旧的app.on_event。在启动时加载模型、初始化信号量和线程池关闭时释放资源。GPU_SEMAPHORE用asyncio.Semaphore保证同时最多4个请求进入推理流程。async with确保异常时也能释放信号量。run_in_executor把同步的model_infer放到线程池执行避免阻塞事件循环。线程池大小和信号量一致防止线程数超过GPU承载能力。torch.cuda.empty_cache()在关闭时清理显存缓存避免残留占用。4.4 压测验证与参数调优代码写完了怎么验证信号量设置是否合理我一般用locust或者wrk做压测。# 安装locust pip install locust # 写一个简单的压测脚本 locustfile.py from locust import HttpUser, task, between class InferUser(HttpUser): wait_time between(0.1, 0.5) task def infer(self): self.client.post(/infer, json{ prompt: test prompt for load testing, max_tokens: 128 })启动压测locust -f locustfile.py --hosthttp://localhost:8000然后在浏览器里设置并发用户数观察响应时间和错误率。调优策略如果显存占用远低于上限可以逐步调大信号量如果出现OOM错误调小信号量。我一般会留20%的显存余量避免突发流量导致溢出。实操心得压测时一定要监控nvidia-smi的显存变化。如果显存曲线呈锯齿状且峰值接近上限说明信号量偏大如果显存平稳但GPU利用率低说明信号量偏小。5. 进阶技巧多卡调度、超时控制与优雅降级5.1 多GPU场景下的请求分发单卡跑满了怎么办上多卡。多卡场景下每个GPU维护自己的信号量和线程池请求进来时根据GPU的当前负载分发。最简单的策略是轮询请求依次分配给GPU0、GPU1、GPU2。但轮询不考虑各卡的实际负载可能导致某张卡排队严重。更好的策略是“最少连接数”每个GPU维护一个计数器记录当前正在处理的请求数新请求分配给计数器最小的GPU。实现上可以用一个共享的字典加锁。class GPUManager: def __init__(self, num_gpus: int, sem_per_gpu: int): self.gpus [] for i in range(num_gpus): self.gpus.append({ id: i, semaphore: asyncio.Semaphore(sem_per_gpu), active: 0, executor: ThreadPoolExecutor(max_workerssem_per_gpu) }) self.lock asyncio.Lock() async def acquire(self): async with self.lock: gpu min(self.gpus, keylambda g: g[active]) gpu[active] 1 await gpu[semaphore].acquire() return gpu async def release(self, gpu): gpu[semaphore].release() async with self.lock: gpu[active] - 1这个实现里active计数在获取信号量之前就增加避免多个请求同时选中同一张卡。释放时先释放信号量再减计数保证顺序正确。5.2 请求超时与取消处理GPU推理可能因为各种原因变慢比如输入特别长、模型遇到复杂样本。如果请求没有超时控制慢请求会一直占着信号量导致后续请求排队。FastAPI本身支持超时中间件但更精细的控制应该在信号量层面做。可以用asyncio.wait_for包裹推理调用async with GPU_SEMAPHORE: try: result await asyncio.wait_for( loop.run_in_executor(INFER_EXECUTOR, model_infer, prompt), timeout30.0 ) except asyncio.TimeoutError: raise HTTPException(status_code504, detailInference timeout)但要注意wait_for超时后线程池里的任务并不会自动取消。如果推理函数不支持中断线程会继续跑直到完成显存也不会立即释放。所以超时时间要设置得比正常推理时间长一些避免误杀。注意Python的线程无法强制终止。如果推理函数卡死只能等它自己结束。生产环境建议用进程池替代线程池超时后直接杀掉进程但进程池的启动开销更大。5.3 显存不足时的优雅降级即使做了并发控制极端情况下显存还是可能不够。比如模型加载后系统其他进程突然占了显存或者CUDA上下文出现碎片。这时候需要优雅降级捕获OOM异常返回友好的错误信息而不是让服务崩溃。app.post(/infer) async def infer(request: InferRequest): try: async with GPU_SEMAPHORE: result await loop.run_in_executor(...) except torch.cuda.OutOfMemoryError: torch.cuda.empty_cache() raise HTTPException( status_code503, detailGPU memory insufficient, please retry later )torch.cuda.empty_cache()会释放未使用的显存缓存有时候能救回来。如果频繁OOM说明信号量设置过大需要调小。另外可以加一个降级策略当显存不足时自动降低批大小或者切换到CPU推理如果模型支持。CPU推理慢但至少服务不挂。6. 常见问题与排查技巧实录6.1 信号量设置了但显存还是溢出这种情况我遇到过几次原因通常有三个原因一信号量是进程内的多worker导致总并发翻倍。如果你用uvicorn --workers 4启动每个worker有自己的信号量总并发是4倍。解决办法是用--workers 1或者把并发控制移到进程外比如用Redis做分布式信号量。原因二推理函数内部还有并发。比如模型内部用了多线程或者CUDA流实际显存占用比预期高。解决办法是限制模型内部的并发设置torch.set_num_threads(1)或者用CUDA_VISIBLE_DEVICES隔离。原因三显存碎片。长时间运行后显存会出现碎片导致明明有足够空闲显存但分配失败。解决办法是定期重启服务或者用torch.cuda.empty_cache()清理。6.2 请求排队时间过长怎么优化排队时间长说明信号量太小或者单次推理太慢。优化方向提升单次推理速度用更快的推理引擎ONNX Runtime、TensorRT、量化模型、减少生成长度。增大信号量如果显存有余量适当调大。动态批处理把多个请求攒成一批提升GPU利用率。加机器单卡到瓶颈了上多卡或者多机。我一般先看GPU利用率。如果利用率低于60%说明信号量偏小或者批处理没做好如果利用率高于90%但排队还是长说明计算量确实大需要加卡或者优化模型。6.3 常见问题速查表问题现象可能原因排查方法解决方案显存溢出并发过高查看nvidia-smi显存曲线调小信号量请求超时推理太慢记录单次推理耗时优化模型或加超时GPU利用率低信号量太小监控GPU利用率调大信号量服务崩溃重启OOM未捕获查看服务日志加OOM异常处理多worker并发翻倍信号量进程内检查worker数量改单worker或分布式信号量6.4 独家避坑技巧技巧一预热。服务启动后先跑几次推理让CUDA上下文和显存分配器预热避免第一个请求特别慢。技巧二监控。用Prometheus加Grafana监控显存、GPU利用率、请求队列长度、推理延迟。这些指标能帮你提前发现问题。技巧三限流。在信号量之前加一层限流比如每秒最多接受100个请求超过的直接返回429。避免请求无限堆积。技巧四日志。记录每个请求的排队时间、推理时间、显存占用。出问题时这些日志是排查的关键。技巧五灰度。调整信号量时不要一次调太多先调10%观察一段时间再继续。7. 我个人在实际操作中的体会这套方案我在三个项目里用过从单卡7B模型到四卡70B模型整体稳定性不错。最大的体会是并发控制不是一劳永逸的需要根据实际负载持续调优。模型换了、请求模式变了、硬件升级了信号量都要重新算。另外别迷信“最优解”。信号量方案简单但够用请求队列复杂但灵活动态批处理强大但难维护。选适合团队当前阶段的方案先跑起来再迭代。最后分享一个小技巧如果你不确定信号量设多少先设1然后压测逐步加。每次加1观察显存和延迟。找到那个“显存快满但还没满”的点就是最优值。这个过程可能花半小时但比上线后炸显存强得多。 FastAPI GPU推理并发控制：信号量与队列解决显存溢出
1. 从一次线上事故说起为什么并发控制是GPU推理服务的生死线去年冬天我帮一个团队排查他们AI绘画服务的线上故障。现象很典型单张图片生成耗时约3秒平时跑得好好的一到晚上高峰期服务就像被掐住脖子一样请求排队越来越长最后直接返回500错误。登录服务器一看nvidia-smi显示显存占用100%但GPU利用率只有30%出头——典型的显存溢出导致的进程崩溃重启循环。这个场景在FastAPI GPU推理的架构里太常见了。FastAPI本身是异步框架天生适合处理高并发IO密集型任务但GPU推理是计算密集型而且显存是独占资源。当你用FastAPI直接暴露一个推理接口每个请求都去加载模型或者分配显存并发一上来显存瞬间被撑爆服务直接挂掉。这篇文章就是围绕这个问题展开的。我会从架构设计、并发控制策略、显存管理、实操代码、踩坑经验几个维度把FastAPI GPU推理并发控制这件事讲透。适合正在用FastAPI做AI服务后端、遇到显存溢出问题的开发者也适合准备把本地推理模型部署成API服务的朋友。读完你至少能搞清楚三件事为什么并发会炸显存、怎么用信号量和队列控制并发、以及怎么在FastAPI里优雅地管理GPU资源。2. 问题根源拆解FastAPI的异步模型与GPU推理的冲突2.1 FastAPI的异步机制到底在异步什么FastAPI基于Starlette和Pydantic核心卖点是async def带来的高并发能力。但很多人对“异步”有误解以为异步就是“同时处理很多事”。准确地说异步是“在等待IO的时候去处理其他请求”。比如数据库查询、HTTP调用、文件读写这些操作CPU大部分时间在等异步可以把这些等待时间利用起来。但GPU推理不一样。当你调用model.generate()或者model.predict()CPU把数据传给GPUGPU开始计算然后CPU等待结果返回。这个等待过程中GPU在满负荷运转显存被模型权重和中间激活值占满。如果你用async def定义推理接口FastAPI会认为这是个异步任务多个请求可以同时进来每个请求都去调用GPU显存瞬间爆炸。我见过太多人这样写app.post(/infer) async def infer(request: InferRequest): result model(request.input) # 同步阻塞调用 return {result: result}这段代码的问题在于async def里面跑了同步阻塞的GPU推理。事件循环被阻塞不说更致命的是多个请求会同时进入这个函数每个都去抢GPU资源。模型可能只加载了一份但推理过程中的中间张量、KV Cache、批处理缓冲区都是按请求分配的并发一高显存直接溢出。2.2 显存溢出的三种典型触发路径根据我处理过的案例显存溢出通常走这三条路第一条模型重复加载。有些实现为了“隔离”每个请求都重新加载模型或者创建新的推理会话。一个7B参数的模型FP16精度大约占14GB显存两个请求同时加载就是28GB一张24GB的卡直接跪。第二条批处理无上限。为了提升吞吐把多个请求攒成一批送进GPU。但批大小没有上限控制显存占用随批大小线性增长。我见过一个服务把batch_size设成64结果单次推理就吃了40GB显存。第三条KV Cache累积。大语言模型推理时KV Cache会随着生成长度增长。如果并发请求多每个请求都维护自己的KV Cache显存占用是并发数乘以单请求Cache大小。并发32路、每路2GB Cache就是64GB远超单卡容量。这三条路径的共同点是显存分配发生在请求处理过程中而不是服务启动时。FastAPI的并发模型让多个请求同时进入处理流程每个请求都去申请显存最终超过物理上限。2.3 为什么不能简单加个锁有人会说那我加个threading.Lock不就行了理论上可以但实际效果很差。锁的粒度如果太大所有请求串行执行GPU利用率上不去吞吐量惨不忍睹。锁的粒度如果太小又起不到保护显存的作用。更合理的做法是在服务层面控制同时进入GPU推理的请求数量让并发数匹配显存容量。这就是并发控制的核心思路。具体来说需要解决三个问题怎么限制并发数、怎么排队、怎么在请求间公平调度。3. 并发控制方案选型从信号量到推理队列3.1 方案一asyncio.Semaphore信号量控制最直接的办法是用asyncio.Semaphore限制同时执行的推理任务数量。信号量的值根据显存容量和单次推理显存占用计算得出。假设一张24GB显存的卡模型权重占14GB系统预留2GB剩余8GB可用于推理中间态。单次推理峰值额外占用2GB那么信号量可以设为4。这样最多4个请求同时进入GPU推理显存占用控制在1424×224GB以内。代码实现大概是这样import asyncio from fastapi import FastAPI app FastAPI() GPU_SEMAPHORE asyncio.Semaphore(4) app.post(/infer) async def infer(request: InferRequest): async with GPU_SEMAPHORE: result await run_in_executor(model, request.input) return {result: result}这里有个关键点run_in_executor把同步的GPU推理放到线程池里执行避免阻塞事件循环。信号量保证同时最多4个任务在跑。这个方案的优点是简单、直观、容易实现。缺点是信号量是进程内的如果用了多个worker进程比如uvicorn --workers 4每个进程有自己的信号量总并发数会翻倍。另外信号量不区分请求优先级先到的先执行后到的干等。3.2 方案二请求队列加消费者模型更工程化的做法是引入一个请求队列推理任务作为消费者从队列里取任务执行。FastAPI接口只负责把请求塞进队列并等待结果不直接碰GPU。这种架构的好处是并发数完全由消费者数量控制和FastAPI的worker数量解耦可以实现优先级队列、超时丢弃、批量合并等高级功能显存管理更可控因为只有消费者进程会碰GPU。实现上可以用asyncio.Queue做进程内队列也可以用Redis、RabbitMQ做跨进程队列。对于单机多卡场景我推荐用进程内队列加多消费者每个消费者绑定一张GPU。import asyncio from concurrent.futures import ThreadPoolExecutor class InferenceQueue: def __init__(self, num_workers2): self.queue asyncio.Queue(maxsize100) self.executor ThreadPoolExecutor(max_workersnum_workers) self.workers [] async def start(self): for _ in range(2): self.workers.append(asyncio.create_task(self.worker())) async def worker(self): while True: request, future await self.queue.get() try: loop asyncio.get_event_loop() result await loop.run_in_executor( self.executor, model_infer, request ) future.set_result(result) except Exception as e: future.set_exception(e) finally: self.queue.task_done()这个方案比信号量复杂但扩展性更好。队列长度可以控制背压消费者数量可以精确匹配显存容量还能做优雅关闭。3.3 方案三动态批处理加显存感知调度如果追求极致吞吐可以考虑动态批处理。核心思路是消费者从队列里取任务时不只取一个而是把当前队列里所有等待的任务攒成一批根据显存余量决定批大小一次性送进GPU。这个方案对显存管理要求最高需要实时监控显存占用动态调整批大小。实现上可以用pynvml读取显存信息结合模型推理的显存profile计算出当前能承受的最大批大小。我实测下来动态批处理在请求量波动大的场景下吞吐提升明显但实现复杂度也高。如果团队没有专门的推理优化经验建议先从信号量或固定消费者队列做起稳定后再考虑动态批处理。3.4 三种方案对比与选型建议方案实现复杂度并发控制精度吞吐量适用场景信号量低中中单进程、请求量稳定请求队列中高中高多worker、需要优先级动态批处理高高高请求波动大、追求吞吐选型建议如果你的服务是单进程部署请求量不大信号量足够用。如果用了多worker或者需要更精细的控制上请求队列。动态批处理适合有专门优化团队、对吞吐有极致要求的场景。4. 实操落地用信号量加线程池实现稳定的GPU推理服务4.1 环境准备与依赖安装先确认基础环境。Python 3.9以上PyTorch或ONNX Runtime按你的模型框架选。FastAPI和Uvicorn是必须的。pip install fastapi uvicorn[standard] pynvml pip install torch torchvision --index-url https://download.pytorch.org/whl/cu118pynvml用来读取显存信息方便计算信号量初始值。如果你用的是NVIDIA显卡驱动版本建议535以上CUDA版本和PyTorch版本要匹配。4.2 显存容量测算与信号量计算这一步是整个方案的基础。你需要搞清楚三个数模型权重显存、单次推理峰值显存、系统预留显存。模型权重显存可以这样估算参数量乘以精度字节数。FP16是2字节FP32是4字节INT8是1字节。比如7B模型FP16就是7×10^9×214GB。单次推理峰值显存包括中间激活值、KV Cache、批处理缓冲区。这个最好用torch.cuda.max_memory_allocated()实测。跑一次推理记录峰值多跑几次取最大值。系统预留显存建议留2-4GB给CUDA上下文、显存碎片、其他进程用。信号量计算公式信号量 floor((总显存 - 模型权重 - 系统预留) / 单次推理峰值)举个例子24GB卡7B模型FP16占14GB系统预留3GB单次推理峰值1.5GB。信号量 floor((24-14-3)/1.5) floor(7/1.5) 4。注意这个计算是保守估计。实际部署时建议先用小信号量跑压测逐步调大观察显存占用和错误率。4.3 完整代码实现与逐行解析下面是一个可直接运行的完整示例。我用了一个模拟的推理函数你替换成自己的模型调用即可。import asyncio import time from concurrent.futures import ThreadPoolExecutor from contextlib import asynccontextmanager import torch from fastapi import FastAPI, HTTPException from pydantic import BaseModel # 全局资源 GPU_SEMAPHORE: asyncio.Semaphore None INFER_EXECUTOR: ThreadPoolExecutor None MODEL None class InferRequest(BaseModel): prompt: str max_tokens: int 128 class InferResponse(BaseModel): result: str latency_ms: float def load_model(): 加载模型实际替换成你的模型加载逻辑 # 这里用sleep模拟加载耗时 time.sleep(2) return {name: mock-model, device: cuda} def model_infer(prompt: str, max_tokens: int) - str: 同步推理函数运行在线程池里 # 模拟GPU推理耗时 time.sleep(0.5) return fgenerated: {prompt[:20]}... asynccontextmanager async def lifespan(app: FastAPI): global GPU_SEMAPHORE, INFER_EXECUTOR, MODEL # 启动时加载模型 MODEL load_model() # 根据显存计算信号量这里硬编码为4 GPU_SEMAPHORE asyncio.Semaphore(4) # 线程池大小等于信号量避免线程堆积 INFER_EXECUTOR ThreadPoolExecutor(max_workers4) yield # 关闭时清理 INFER_EXECUTOR.shutdown(waitTrue) torch.cuda.empty_cache() app FastAPI(lifespanlifespan) app.post(/infer, response_modelInferResponse) async def infer(request: InferRequest): start time.time() try: async with GPU_SEMAPHORE: loop asyncio.get_event_loop() result await loop.run_in_executor( INFER_EXECUTOR, model_infer, request.prompt, request.max_tokens ) except Exception as e: raise HTTPException(status_code500, detailstr(e)) latency (time.time() - start) * 1000 return InferResponse(resultresult, latency_mslatency) app.get(/health) async def health(): return {status: ok, gpu_available: torch.cuda.is_available()}逐行解析几个关键点lifespan是FastAPI推荐的生命周期管理方式替代了旧的app.on_event。在启动时加载模型、初始化信号量和线程池关闭时释放资源。GPU_SEMAPHORE用asyncio.Semaphore保证同时最多4个请求进入推理流程。async with确保异常时也能释放信号量。run_in_executor把同步的model_infer放到线程池执行避免阻塞事件循环。线程池大小和信号量一致防止线程数超过GPU承载能力。torch.cuda.empty_cache()在关闭时清理显存缓存避免残留占用。4.4 压测验证与参数调优代码写完了怎么验证信号量设置是否合理我一般用locust或者wrk做压测。# 安装locust pip install locust # 写一个简单的压测脚本 locustfile.py from locust import HttpUser, task, between class InferUser(HttpUser): wait_time between(0.1, 0.5) task def infer(self): self.client.post(/infer, json{ prompt: test prompt for load testing, max_tokens: 128 })启动压测locust -f locustfile.py --hosthttp://localhost:8000然后在浏览器里设置并发用户数观察响应时间和错误率。调优策略如果显存占用远低于上限可以逐步调大信号量如果出现OOM错误调小信号量。我一般会留20%的显存余量避免突发流量导致溢出。实操心得压测时一定要监控nvidia-smi的显存变化。如果显存曲线呈锯齿状且峰值接近上限说明信号量偏大如果显存平稳但GPU利用率低说明信号量偏小。5. 进阶技巧多卡调度、超时控制与优雅降级5.1 多GPU场景下的请求分发单卡跑满了怎么办上多卡。多卡场景下每个GPU维护自己的信号量和线程池请求进来时根据GPU的当前负载分发。最简单的策略是轮询请求依次分配给GPU0、GPU1、GPU2。但轮询不考虑各卡的实际负载可能导致某张卡排队严重。更好的策略是“最少连接数”每个GPU维护一个计数器记录当前正在处理的请求数新请求分配给计数器最小的GPU。实现上可以用一个共享的字典加锁。class GPUManager: def __init__(self, num_gpus: int, sem_per_gpu: int): self.gpus [] for i in range(num_gpus): self.gpus.append({ id: i, semaphore: asyncio.Semaphore(sem_per_gpu), active: 0, executor: ThreadPoolExecutor(max_workerssem_per_gpu) }) self.lock asyncio.Lock() async def acquire(self): async with self.lock: gpu min(self.gpus, keylambda g: g[active]) gpu[active] 1 await gpu[semaphore].acquire() return gpu async def release(self, gpu): gpu[semaphore].release() async with self.lock: gpu[active] - 1这个实现里active计数在获取信号量之前就增加避免多个请求同时选中同一张卡。释放时先释放信号量再减计数保证顺序正确。5.2 请求超时与取消处理GPU推理可能因为各种原因变慢比如输入特别长、模型遇到复杂样本。如果请求没有超时控制慢请求会一直占着信号量导致后续请求排队。FastAPI本身支持超时中间件但更精细的控制应该在信号量层面做。可以用asyncio.wait_for包裹推理调用async with GPU_SEMAPHORE: try: result await asyncio.wait_for( loop.run_in_executor(INFER_EXECUTOR, model_infer, prompt), timeout30.0 ) except asyncio.TimeoutError: raise HTTPException(status_code504, detailInference timeout)但要注意wait_for超时后线程池里的任务并不会自动取消。如果推理函数不支持中断线程会继续跑直到完成显存也不会立即释放。所以超时时间要设置得比正常推理时间长一些避免误杀。注意Python的线程无法强制终止。如果推理函数卡死只能等它自己结束。生产环境建议用进程池替代线程池超时后直接杀掉进程但进程池的启动开销更大。5.3 显存不足时的优雅降级即使做了并发控制极端情况下显存还是可能不够。比如模型加载后系统其他进程突然占了显存或者CUDA上下文出现碎片。这时候需要优雅降级捕获OOM异常返回友好的错误信息而不是让服务崩溃。app.post(/infer) async def infer(request: InferRequest): try: async with GPU_SEMAPHORE: result await loop.run_in_executor(...) except torch.cuda.OutOfMemoryError: torch.cuda.empty_cache() raise HTTPException( status_code503, detailGPU memory insufficient, please retry later )torch.cuda.empty_cache()会释放未使用的显存缓存有时候能救回来。如果频繁OOM说明信号量设置过大需要调小。另外可以加一个降级策略当显存不足时自动降低批大小或者切换到CPU推理如果模型支持。CPU推理慢但至少服务不挂。6. 常见问题与排查技巧实录6.1 信号量设置了但显存还是溢出这种情况我遇到过几次原因通常有三个原因一信号量是进程内的多worker导致总并发翻倍。如果你用uvicorn --workers 4启动每个worker有自己的信号量总并发是4倍。解决办法是用--workers 1或者把并发控制移到进程外比如用Redis做分布式信号量。原因二推理函数内部还有并发。比如模型内部用了多线程或者CUDA流实际显存占用比预期高。解决办法是限制模型内部的并发设置torch.set_num_threads(1)或者用CUDA_VISIBLE_DEVICES隔离。原因三显存碎片。长时间运行后显存会出现碎片导致明明有足够空闲显存但分配失败。解决办法是定期重启服务或者用torch.cuda.empty_cache()清理。6.2 请求排队时间过长怎么优化排队时间长说明信号量太小或者单次推理太慢。优化方向提升单次推理速度用更快的推理引擎ONNX Runtime、TensorRT、量化模型、减少生成长度。增大信号量如果显存有余量适当调大。动态批处理把多个请求攒成一批提升GPU利用率。加机器单卡到瓶颈了上多卡或者多机。我一般先看GPU利用率。如果利用率低于60%说明信号量偏小或者批处理没做好如果利用率高于90%但排队还是长说明计算量确实大需要加卡或者优化模型。6.3 常见问题速查表问题现象可能原因排查方法解决方案显存溢出并发过高查看nvidia-smi显存曲线调小信号量请求超时推理太慢记录单次推理耗时优化模型或加超时GPU利用率低信号量太小监控GPU利用率调大信号量服务崩溃重启OOM未捕获查看服务日志加OOM异常处理多worker并发翻倍信号量进程内检查worker数量改单worker或分布式信号量6.4 独家避坑技巧技巧一预热。服务启动后先跑几次推理让CUDA上下文和显存分配器预热避免第一个请求特别慢。技巧二监控。用Prometheus加Grafana监控显存、GPU利用率、请求队列长度、推理延迟。这些指标能帮你提前发现问题。技巧三限流。在信号量之前加一层限流比如每秒最多接受100个请求超过的直接返回429。避免请求无限堆积。技巧四日志。记录每个请求的排队时间、推理时间、显存占用。出问题时这些日志是排查的关键。技巧五灰度。调整信号量时不要一次调太多先调10%观察一段时间再继续。7. 我个人在实际操作中的体会这套方案我在三个项目里用过从单卡7B模型到四卡70B模型整体稳定性不错。最大的体会是并发控制不是一劳永逸的需要根据实际负载持续调优。模型换了、请求模式变了、硬件升级了信号量都要重新算。另外别迷信“最优解”。信号量方案简单但够用请求队列复杂但灵活动态批处理强大但难维护。选适合团队当前阶段的方案先跑起来再迭代。最后分享一个小技巧如果你不确定信号量设多少先设1然后压测逐步加。每次加1观察显存和延迟。找到那个“显存快满但还没满”的点就是最优值。这个过程可能花半小时但比上线后炸显存强得多。