多个人同时提问但位置有限

摘要：在算力昂贵的大模型（LLM）推理与高并发后端架构中，资源的物理约束是不可跨越的红线。假设系统只有 3 个 GPU 槽位（Inference Slots / Worker Threads），当 10 个并发请求同时涌入时，剩余的 7 个请求应该何去何从？

如果不做控制盲目放行，将引发显存溢出（OOM）、推理性能雪崩与连接挂死；如果简单粗暴全部拒绝，又会带来极差的用户体验。

本文将以“10人抢3坑”为经典场景，深入剖析并发槽位控制的底层逻辑，系统拆解排队等待（Queuing）、自回归连续批处理（Continuous Batching）、背压限流（Backpressure）与优雅降级（Degradation）四大核心方案，并提供一套可直接用于生产的 Python / FastAPI 动态排队与槽位调度全套实战代码。

前言：算力物理极限下的“并发困局”

无论后台采用多么强大的集群（如 H100、A100 或国产算力卡），系统的并发处理能力永远是有上限的。

在传统 Web 业务中，处理一个 HTTP 请求通常只需 10~50 毫秒的 CPU 周期；但在大语言模型（LLM）时代，一个包含长文本推理和逐字生成的请求，往往需要持续占用 GPU 算力3 秒到 30 秒。

这就引出了一个非常经典且严峻的工程架构问题：

“系统只有 3 个可用的推理槽位（Slot / Worker），当 10 个用户在同一秒内发起提问时，系统该如何保证高可用、低延迟与公平性？”

┌── 用户 1 (执行中) ────► [ 物理槽位 1 ] ├── 用户 2 (执行中) ────► [ 物理槽位 2 ] ├── 用户 3 (执行中) ────► [ 物理槽位 3 ] │ 10 个并发请求 ─┼── 用户 4 ──┐ ├── 用户 5 │ ├── 用户 6 │ ├── 用户 7 ├────────► 【 调度器 / 排队等待队列 】 ├── 用户 8 │ (如何管理？如何通知？何时超时？) ├── 用户 9 │ └── 用户 10 ─┘

如果架构设计不当，可能会引发三大系统灾难：

1. 显存击穿与 OOM 崩溃：若不做限制强行启动 10 个并发，动态膨胀的 KV Cache 将瞬间撑爆 GPU 显存，导致推理服务整体宕机。

2. 延迟雪崩与吞吐腰斩：10 个请求互相争抢有限的显存带宽与算力核心，导致所有用户的首字时间（TTFT）从 300 毫秒飙升到 20 秒，全员体验崩溃。

3. 僵尸请求积压：排队的用户早已关闭了浏览器网页，后端却依然在消耗昂贵的算力生成无用的答案。

一、 问题的本质：什么是“坑位”（Slot）？

在深入调度算法之前，我们先在概念上搞清楚：“坑位”究竟代表什么？

1.1 从传统线程池到 GPU 推理槽位

不同技术栈对“坑位”的定义有所不同：

• 传统后端（Java/Go/C++）：坑位代表工作线程数（Worker Threads）或数据库连接池（Connection Pool）大小。

• 本地模型部署（如 Ollama）：坑位代表并发上下文实例参数OLLAMA_NUM_PARALLEL。如果设为 3，同一时刻只能有 3 个模型推理实例活跃。

• 专业推理引擎（vLLM / TensorRT-LLM）：坑位代表最大并发序列数（Max Num Sequences）与显存分页块（PagedAttention Blocks）的容量上限。

1.2 盲目并发 vs 受控调度的物理表现

我们可以通过下表对比在算力受限时，是否做槽位控制的巨大差异：

二、 四大应对哲学：10 人抢 3 坑的技术路线演进

当 10 个人争抢 3 个坑位时，业界主要演进出四种处理路线：

┌── 策略 1: 快速失败 (Fast-Fail / 429 拒载) │ ├── 策略 2: 异步排队 (FIFO / 优先级队列 + 状态推送) 四大核心策略 ──┤ ├── 策略 3: 连续批处理 (Continuous Batching / 细粒度调度) │ └── 策略 4: 降级分流 (Model Routing / 备用小模型兜底)

2.1 策略一：快速失败与背压（Fast-Fail & Backpressure）

• 核心逻辑：如果 3 个槽位全满，剩余 7 个请求中，如果等待队列也达到上限，系统立即拒绝多余请求，返回 HTTP429 Too Many Requests或503 Service Unavailable。

• 优点：架构最简单，绝对保护后端核心系统不被冲垮。

• 缺点：用户体验较生硬，直接报错可能导致用户流失。

• 最佳实践：在响应头中附带Retry-After: 5，告诉客户端 5 秒后再试，配合前端做自动抖动重试。

2.2 策略二：排队缓冲与状态感知（Async Queuing & ETA）

• 核心逻辑：3 个请求进槽位执行，剩余 7 个请求进入内存等待队列（FIFO 或 Priority Queue）。

• 体验增强：通过长连接（SSE / WebSocket）向排队中的 7 个客户端实时推送排队顺位与预估等待时间。

• 排队论与预估时间（ETA）公式：

  根据排队论（Little's Law），排队等待时间取决于队列位置、当前槽位数量与历史平均推理耗时：

  预估等待时间 (ETA) = ( 当前排队位置 / 活跃槽位数 ) * 历史平均请求处理耗时

  例如：用户小明排在第 4 位，槽位数为 3，平均每个请求耗时 6 秒，则小明的预估等待时间约为：( 4 / 3 ) * 6 秒 = 8 秒。

2.3 策略三：大模型专属的连续批处理（Continuous Batching）

在 vLLM、TGI 等现代化 LLM 推理引擎中，“槽位”并不是绝对固定的物理栅栏，而是通过Iteration-level 细粒度调度实现了动态槽位流转：

时间步 1: [槽位 1: 请求 A] [槽位 2: 请求 B] [槽位 3: 请求 C] (队列待命: D, E, F...) 时间步 2: [槽位 1: 请求 A] [槽位 2: 请求 B (提前结束)] ➔ 槽位 2 释放! 时间步 3: [槽位 1: 请求 A] [槽位 2: 请求 D (立即补位!)] [槽位 3: 请求 C]

• 原理解析：当请求 B 只需要生成 20 个字并提前遇到结束符时，系统无需等待 A 和 C 完成，在下一个时间步（Next-Token 迭代）立刻把排队的请求 D 塞进槽位 2。

• 效果：将 3 个固定槽位的整体流转效率提升 300% 以上。

2.4 策略四：智能模型路由与降级（Model Routing & Fallback）

• 核心逻辑：如果高精度的主模型（如 70B 模型）3 个槽位已满，调度网关通过意图识别，将排队中的简单请求（如日常问候、短文润色）自动路由至轻量级备用小模型（如 7B/8B 模型或云端 Serverless API）。

• 效果：实现流量分流，兼顾响应速度与成本。

三、 架构设计：高可用并发槽位调度网关

为了在生产环境中完美解决“10人抢3坑”，我们需要设计一个集成了信号量控制、异步等待队列、SSE 顺位推送、断连探测与超时丢弃的统一网关。

3.1 架构拓扑图

┌────────────────────────────────────────────────────────────────────────┐ │ 客户端层 (10 个并发请求) │ └───────────────────────────────────┬────────────────────────────────────┘ │ HTTP POST (SSE 流式连接) ▼ ┌────────────────────────────────────────────────────────────────────────┐ │ 调度网关层 (FastAPI / Web Gateway) │ │ │ │ 1. 准入控制器 (Admission Control) ➔ 校验系统总负载 │ │ 2. 异步信号量 (Async Semaphore, 容量=3) ➔ 物理槽位门禁 │ │ 3. 动态排队管理器 (Queue Manager, 最大等待深度=20) │ │ 4. 客户端存活探测器 (Disconnect Listener) ➔ 拦截取消事件 │ └───────────────────────────────────┬────────────────────────────────────┘ │ 获得槽位 (Acquire Token) ▼ ┌────────────────────────────────────────────────────────────────────────┐ │ LLM 推理执行引擎 (vLLM / Worker Pool) │ │ [ 槽位 1 ] [ 槽位 2 ] [ 槽位 3 ] │ └────────────────────────────────────────────────────────────────────────┘

四、 生产级代码实战：带排队感知与槽位调度的 Python 网关

下面给出一份完整的生产级 Python 实现代码，基于FastAPI、asyncio.Semaphore和Server-Sent Events (SSE)。

4.1 核心代码实现

import asyncio import json import time import uuid from typing import AsyncGenerator, Dict from fastapi import FastAPI, Request, HTTPException from fastapi.responses import StreamingResponse from pydantic import BaseModel app = FastAPI(title="LLM Concurrency & Slot Dispatcher Gateway") # ==================== 1. 核心调度参数配置 ==================== MAX_ACTIVE_SLOTS = 3 # 物理并发槽位总数 (3个坑位) MAX_QUEUE_SIZE = 10 # 最大排队等待深度 (超出直接快速失败) QUEUE_TIMEOUT_SECONDS = 60 # 排队最大超时时间 (秒) AVG_PROCESS_TIME = 5.0 # 历史平均单请求耗时 (秒, 用于预估时间) # 全局并发控制器 slot_semaphore = asyncio.Semaphore(MAX_ACTIVE_SLOTS) active_tasks_count = 0 # 当前正在占用槽位的任务数 waiting_queue = [] # 等待队列，存放 task_id lock = asyncio.Lock() # 保护队列状态的互斥锁 class ChatRequest(BaseModel): prompt: str user_id: str = "anonymous" class TaskContext: def __init__(self, task_id: str, prompt: str): self.task_id = task_id self.prompt = prompt self.enqueue_time = time.time() self.acquired_event = asyncio.Event() # 当成功获得槽位时 set() self.is_cancelled = False # ==================== 2. 模拟 LLM 真实推理引擎 ==================== async def mock_llm_inference(prompt: str, task_id: str) -> AsyncGenerator[str, None]: """模拟大模型逐字生成 (Decode 阶段耗时)""" tokens = [f"【槽位处理】", f"针对", f"输入:「{prompt}」", f"，模型", f"开始", f"生成", f"答案...", f"完成!"] for token in tokens: await asyncio.sleep(0.6) # 模拟单步计算耗时 chunk_data = { "task_id": task_id, "token": token, "status": "generating" } yield f"data: {json.dumps(chunk_data, ensure_ascii=False)}\n\n" yield "data: [DONE]\n\n" # ==================== 3. 排队管理与槽位调度核心逻辑 ==================== async def queue_and_execute_stream(task: TaskContext, raw_request: Request) -> AsyncGenerator[str, None]: global active_tasks_count # A. 准入与排队判定 async with lock: if len(waiting_queue) >= MAX_QUEUE_SIZE: # 超过最大排队深度，立即触发 Fast-Fail 背压拒绝 error_payload = {"error": "Server is too busy. Queue is full.", "code": 429} yield f"data: {json.dumps(error_payload, ensure_ascii=False)}\n\n" return waiting_queue.append(task) current_pos = len(waiting_queue) try: # B. 排队轮询与状态实时推送循环 print(f"--> 任务 [{task.task_id}] 进入等待队列，当前排在第 {current_pos} 位") while True: # 1. 探测客户端是否已提前断开连接（如用户关闭了网页） if await raw_request.is_disconnected(): print(f"xx 任务 [{task.task_id}] 客户端已断开，取消排队并释放资源") task.is_cancelled = True return # 2. 检查排队是否超时 if time.time() - task.enqueue_time > QUEUE_TIMEOUT_SECONDS: timeout_payload = {"error": "Queue timeout. Please try again later.", "code": 504} yield f"data: {json.dumps(timeout_payload, ensure_ascii=False)}\n\n" return # 3. 尝试以非阻塞方式探测信号量 (抢占槽位) # 如果当前任务排在队列第一位，且有空余槽位，则竞争槽位 async with lock: if waiting_queue and waiting_queue[0].task_id == task.task_id: if not slot_semaphore.locked(): await slot_semaphore.acquire() waiting_queue.pop(0) # 移出等待队列 active_tasks_count += 1 print(f"==> 任务 [{task.task_id}] 成功获取物理槽位！当前活跃槽位: {active_tasks_count}/{MAX_ACTIVE_SLOTS}") break # 计算当前顺位与 ETA try: current_idx = [t.task_id for t in waiting_queue].index(task.task_id) + 1 except ValueError: current_idx = 1 # 4. 向前端推送排队状态帧 (SSE) estimated_wait_time = round((current_idx / MAX_ACTIVE_SLOTS) * AVG_PROCESS_TIME, 1) status_payload = { "task_id": task.task_id, "status": "queuing", "queue_position": current_idx, "estimated_wait_seconds": estimated_wait_time } yield f"event: queue_status\ndata: {json.dumps(status_payload, ensure_ascii=False)}\n\n" # 每隔 1 秒自旋轮询一次 await asyncio.sleep(1.0) # C. 槽位执行阶段 (已成功获得信号量) try: # 通知前端排队结束，开始接收文本 start_payload = {"task_id": task.task_id, "status": "started"} yield f"event: start\ndata: {json.dumps(start_payload, ensure_ascii=False)}\n\n" # 真正执行推理 async for token_msg in mock_llm_inference(task.prompt, task.task_id): # 运行中持续探测连接是否存活 if await raw_request.is_disconnected(): print(f"xx 任务 [{task.task_id}] 执行过程中客户端断开连接，立即中断模型生成！") break yield token_msg finally: # D. 核心安全保障：无论成功、异常还是客户端主动断开，必须强制释放槽位 async with lock: active_tasks_count -= 1 slot_semaphore.release() print(f"<-- 任务 [{task.task_id}] 执行完毕退出，归还槽位。剩余活跃槽位: {active_tasks_count}/{MAX_ACTIVE_SLOTS}") finally: # 清理异常情况下仍在队列中的任务 async with lock: if task in waiting_queue: waiting_queue.remove(task) # ==================== 4. HTTP API 入口定义 ==================== @app.post("/v1/chat/completions") async def chat_stream_endpoint(req: ChatRequest, raw_request: Request): task_id = str(uuid.uuid4())[:8] task_ctx = TaskContext(task_id=task_id, prompt=req.prompt) return StreamingResponse( queue_and_execute_stream(task_ctx, raw_request), media_type="text/event-stream", headers={ "Cache-Control": "no-cache", "Connection": "keep-alive", "X-Accel-Buffering": "no" # 禁用 Nginx 缓冲 } ) if __name__ == "__main__": import uvicorn # 启动单进程异步服务进行压力验证 uvicorn.run(app, host="0.0.0.0", port=8000)

4.2 为什么必须使用finally强制释放？

在上述代码中，槽位释放逻辑必须放置在try...finally代码块内。

在大模型生产环境中，客户端随时可能发生异常情况：

• 用户点击“停止生成”按钮；

• 用户直接关闭浏览器标签页；

• 客户端弱网发生 TCP RST 断连。

如果没有finally保障，一旦客户端断开，协程可能在半途被销毁，导致信号量（Semaphore）永远无法归还，形成“僵尸槽位泄露（Zombie Slot Leak）”。只需发生 3 次异常，整个系统就会彻底挂死，后续所有请求全被永久阻塞！

五、 前端与用户体验设计：如何让等待的 7 个人不感到焦虑？

在 10 个人争抢 3 个槽位的场景下，纯后端的控制只完成了一半工作，前端交互的心理学设计决定了最终的用户满意度。

5.1 拒绝对白屏的盲目等待：可视化排队面板

前端通过监听 SSE 推送的event: queue_status事件，在界面上动态展示排队进度：

┌────────────────────────────────────────────────────────┐ │ AI 助手正在全速运算中... │ ├────────────────────────────────────────────────────────┤ │ │ │ 当前前方排队人数: 3 人 │ │ 您的排队顺位: [ 第 4 位 ] │ │ 预估等待时间: 约 8 秒 │ │ │ │ [██████████████░░░░░░░░░░░░░░░░░░░░] 40% │ │ │ │ [ 取消提问 ] │ └────────────────────────────────────────────────────────┘

5.2 前端核心处理逻辑（JavaScript 示例）

const response = await fetch("/v1/chat/completions", { method: "POST", headers: { "Content-Type": "application/json" }, body: JSON.stringify({ prompt: "帮我写一篇关于高并发调度的博客" }) }); const reader = response.body.getReader(); const decoder = new TextDecoder("utf-8"); while (true) { const { value, done } = await reader.read(); if (done) break; const chunk = decoder.decode(value, { stream: true }); // 监听排队状态事件 if (chunk.includes("event: queue_status")) { const dataMatch = chunk.match(/data:\s*(.*)/); if (dataMatch) { const statusData = JSON.parse(dataMatch[1]); updateQueueUI(statusData.queue_position, statusData.estimated_wait_seconds); } } // 监听正式生成事件 else if (chunk.includes("status") && chunk.includes("generating")) { hideQueueUI(); renderTokenToScreen(chunk); } }

六、 生产环境避坑指南与容量规划

在大规模集群部署中，调度机制还需要考虑分布式协作与高可用治理。

6.1 避坑一：惊群效应（Thundering Herd）

当 3 个槽位中的某一个槽位释放时，如果采用无序广播通知等待队列中的 7 个人同时去抢锁，会导致 CPU 瞬间出现剧烈的自旋争抢与上下文切换。

• 解法：严格使用FIFO 有序通知队列（如 Python 的asyncio.Queue或 Redis 的有序集合 ZSet），每次仅精准唤醒排在队首的第 1 个任务，其余任务保持静默挂起。

6.2 避坑二：分布式环境下的集中式槽位控制

当网关部署了多个 Pod 实例（如 Kubernetes 集群部署 5 个 Gateway 副本）时，内存级的asyncio.Semaphore无法跨实例共享。

• 解法：引入Redis + Lua 脚本实现分布式分布式信号量与排队管理器，或者在推理集群前置专业的 AI 网关（如 Cloudflare AI Gateway、Envoy AI Gateway）。

6.3 监控指标与报警基线（Prometheus Metrics）

生产级槽位调度系统必须暴露以下关键指标大盘：

1. llm_active_slots_gauge (当前活跃占用的槽位数，阈值 > 90% 触发报警) 2. llm_queue_depth_gauge (当前排队等待的请求数) 3. llm_queue_wait_time_seconds (排队等待耗时直方图，P95/P99) 4. llm_slot_holding_time_seconds (单次槽位占用时长分布) 5. llm_rejection_total (因排队超时或队列满被丢弃的 429 请求总数)

结语

在资源有限的真实世界中，“10 个人抢 3 个坑位”不是偶发的异常，而是所有成功系统必须面对的常态化挑战。

解决这一问题的技术演进展现了严密的架构工程思维：

1. 基础防线：用信号量（Semaphore）设立刚性物理红线，绝对杜绝显存 OOM 与性能崩溃；

2. 缓冲机制：用有序队列（Queue）与排队状态推送抚平流量洪峰，将生硬的拒绝转化为确定性的等待体验；

3. 精细运营：用连接探测（Cancellation Detection）杜绝算力浪费，用连续批处理（Continuous Batching）榨干每一毫秒的算力潜能。

掌握高并发槽位调度与背压设计，才能让系统在汹涌的流量洪峰面前，始终保持沉着、优雅与坚固。