模型推理的请求调度优先级队列与公平性保障的算法设计一、引言为什么VIP用户的推理请求要等30秒而免费用户的请求反而秒回这是我们多租户推理平台上线第二周产品经理甩到我面前的数据。查了一圈后发现调度器用的是最朴素的FIFO队列。高峰时段来自实时对话系统的高优请求和离线的批量评测请求混在一起排队先到先服务——这对付费客户而言显然不可接受。这个问题的本质是在多租户共享GPU资源的前提下如何在有限算力中实现差异化服务的同时防止低优先级请求被饿死单纯的FIFO做不到纯粹的优先级队列能做到差异化但会导致饥饿而加权的公平队列WFQ又无法满足SLO驱动的动态调整需求。经过三版迭代我们最终落地了一套混合调度方案令牌桶做入口限流 → 三级优先级分类 → 优先级队列调度 → SLO驱动的动态权重调整 → 反饥饿监控。这篇文章完整呈现这个算法的设计思路、核心代码和边界条件。二、原理剖析混合调度算法架构2.1 整体调度流水线graph TB A[推理请求到达] -- B[请求分类器br/PriorityClassifier] B -- C{优先级判定} C --|P0-实时| D[高优令牌桶br/rate200/s] C --|P1-交互| E[中优令牌桶br/rate100/s] C --|P2-批量| F[低优令牌桶br/rate50/s] D -- G{令牌可用?} E -- G F -- G G --|否| H[429 Too Many Requestsbr/触发降级或排队] G --|是| I[入队br/PriorityQueue] I -- J[调度器br/Scheduler Loop] J -- K{选择下一个请求} K -- L[动态权重计算br/SLO-Driven Weight] L -- M[出队发送到br/推理引擎] M -- N[推理执行br/vLLM/TGI] N -- O[结果返回] subgraph 反饥饿机制 P[Starvation Monitorbr/每10s检查] P -- Q{低优等待 60s?} Q --|是| R[临时提升优先级br/Boost Duration30s] Q --|否| S[正常调度] end J -.- P style I fill:#FFD700,stroke:#333,stroke-width:2px style L fill:#87CEEB,stroke:#333,stroke-width:2px style P fill:#FF6B6B,stroke:#333,stroke-width:2px2.2 优先级分类策略我们在网关层根据请求来源和业务属性做三级分类优先级标签典型场景SLO要求权重基数P0实时/关键路径对话生成、代码补全、搜索摘要TTFT 300ms, P9970%P1交互/非关键文档润色、翻译、数据分析TTFT 1s, P9925%P2批量/离线数据集评测、Embedding生成、离线推理无硬性SLO尽力而为5%关键设计点权重是动态的而非静态的。当P0队列为空时P1自动获得更多调度机会当P1的SLO出现恶化趋势时它的权重会被临时上浮。2.3 令牌桶入口限流为什么需要令牌桶因为即使有优先级队列如果P0请求无限制涌入GPU资源的物理上限决定了P1/P2仍然得不到服务。令牌桶的作用是在入口处对各级别做速率限制保证低优先级请求有一个保底的GPU时间份额。graph LR subgraph 令牌桶算法 T[令牌生成br/rate200/s] -- B[(桶容量br/burst400)] B -- C{请求到达} C --|令牌足够| D[放行] C --|令牌不足| E[限流拒绝] end subgraph 三级限流效果 F[P0: 200 req/s] -- G[GPU分配 ≈70%] H[P1: 100 req/s] -- G[GPU分配 ≈25%] I[P2: 50 req/s] -- G[GPU分配 ≈5%] end2.4 SLO驱动的动态权重调整这是整个调度算法最核心的部分。我们监控每个优先级近5分钟的SLO达成率当某个优先级的SLO达标率跌破目标如P0的TTFT P99 300ms的达标率低于99%自动上浮其调度权重新权重 基础权重 × (1 SLO偏离系数) SLO偏离系数 max(0, 目标达标率 - 实际达标率) × 放大因子这确保在系统负载异常升高时高优先级的服务体验得到优先保障。三、生产级代码实现3.1 请求分类器public enum RequestPriority { P0_REALTIME(0, 0.70), // 实时 P1_INTERACTIVE(1, 0.25), // 交互 P2_BATCH(2, 0.05); // 批量 public final int level; public final double baseWeight; RequestPriority(int level, double baseWeight) { this.level level; this.baseWeight baseWeight; } } public class PriorityClassifier { /** * 根据请求来源和业务类型判定优先级 */ public static RequestPriority classify(InferenceRequest request) { // P0: 实时对话、代码补全用户直接等待 if (request.getSource() Source.REALTIME_CHAT || request.getSource() Source.CODE_COMPLETION) { return RequestPriority.P0_REALTIME; } // P1: 交互式但非实时文档处理、翻译 if (request.getSource() Source.DOC_PROCESSOR || request.getSource() Source.TRANSLATION) { return RequestPriority.P1_INTERACTIVE; } // P2: 离线批量 return RequestPriority.P2_BATCH; } }3.2 令牌桶实现public class RateLimiter { private final AtomicLong availableTokens; private final long ratePerSecond; private final long maxBurst; private volatile long lastRefillTime; public RateLimiter(long ratePerSecond, long maxBurst) { this.ratePerSecond ratePerSecond; this.maxBurst maxBurst; this.availableTokens new AtomicLong(maxBurst); this.lastRefillTime System.nanoTime(); } /** * 尝试获取一个令牌非阻塞 * return true放行, false限流 */ public boolean tryAcquire() { refill(); while (true) { long current availableTokens.get(); if (current 0) return false; if (availableTokens.compareAndSet(current, current - 1)) { return true; } } } private void refill() { long now System.nanoTime(); long elapsedNanos now - lastRefillTime; long newTokens elapsedNanos * ratePerSecond / 1_000_000_000L; if (newTokens 0) { availableTokens.updateAndGet(current - Math.min(maxBurst, current newTokens)); lastRefillTime now; } } }3.3 优先级调度器核心public class PriorityScheduler { // 三级优先级队列每级独立 private final MapRequestPriority, BlockingQueueScheduledRequest queues; // 各优先级的令牌桶 private final MapRequestPriority, RateLimiter limiters; // 动态权重原子更新 private final AtomicReferenceMapRequestPriority, Double dynamicWeights; // 饥饿监控 private final StarvationMonitor starvationMonitor; public PriorityScheduler() { this.queues new EnumMap(RequestPriority.class); this.limiters new EnumMap(RequestPriority.class); // 初始化队列和限流器 for (RequestPriority p : RequestPriority.values()) { queues.put(p, new LinkedBlockingQueue(5000)); } limiters.put(RequestPriority.P0_REALTIME, new RateLimiter(200, 400)); limiters.put(RequestPriority.P1_INTERACTIVE, new RateLimiter(100, 200)); limiters.put(RequestPriority.P2_BATCH, new RateLimiter(50, 100)); // 初始权重 基础权重 MapRequestPriority, Double initialWeights new EnumMap(RequestPriority.class); for (RequestPriority p : RequestPriority.values()) { initialWeights.put(p, p.baseWeight); } this.dynamicWeights new AtomicReference(initialWeights); this.starvationMonitor new StarvationMonitor(queues); } /** * 提交请求到调度队列 */ public boolean submit(InferenceRequest request) { RequestPriority priority PriorityClassifier.classify(request); // 入口限流 if (!limiters.get(priority).tryAcquire()) { // P2可降级排队P0/P1拒绝 if (priority RequestPriority.P2_BATCH) { // 返回排队令牌给客户端 return false; } throw new RateLimitExceededException( Priority priority rate limit exceeded); } ScheduledRequest scheduled new ScheduledRequest(request, priority); return queues.get(priority).offer(scheduled); } /** * 调度循环运行在独立线程上 */ public void scheduleLoop() { while (!Thread.currentThread().isInterrupted()) { ScheduledRequest next selectNextRequest(); if (next ! null) { dispatchToEngine(next); } else { // 所有队列为空短暂sleep LockSupport.parkNanos(TimeUnit.MILLISECONDS.toNanos(1)); } } } /** * 加权选择下一个请求 * * 核心算法基于动态权重的概率选择 * - 生成[0, totalWeight)的随机数 * - 按优先级从高到低遍历落在谁的区间就选谁 * - 如果该优先级队列为空跳过并重新分配权重 */ private ScheduledRequest selectNextRequest() { MapRequestPriority, Double weights dynamicWeights.get(); // 过滤空队列 ListRequestPriority available Arrays.stream(RequestPriority.values()) .filter(p - !queues.get(p).isEmpty()) .toList(); if (available.isEmpty()) return null; // 归一化权重 double totalWeight available.stream() .mapToDouble(weights::get).sum(); double random ThreadLocalRandom.current().nextDouble() * totalWeight; double cumulative 0.0; for (RequestPriority p : available) { cumulative weights.get(p); if (random cumulative) { ScheduledRequest req queues.get(p).poll(); if (req ! null) { // 更新饥饿监控 starvationMonitor.recordDispatch(p); return req; } } } // Fallback取最高优先级队列 for (RequestPriority p : RequestPriority.values()) { ScheduledRequest req queues.get(p).poll(); if (req ! null) return req; } return null; } }3.4 SLO驱动的动态权重调整public class SloDrivenWeightAdjuster implements Runnable { private final PriorityScheduler scheduler; private final MetricsStore metricsStore; private static final double AMPLIFICATION_FACTOR 5.0; private static final long ADJUST_INTERVAL_SECONDS 30; Override public void run() { while (!Thread.currentThread().isInterrupted()) { try { adjustWeights(); Thread.sleep(ADJUST_INTERVAL_SECONDS * 1000); } catch (InterruptedException e) { Thread.currentThread().interrupt(); break; } } } private void adjustWeights() { MapRequestPriority, Double newWeights new EnumMap(RequestPriority.class); for (RequestPriority p : RequestPriority.values()) { double baseWeight p.baseWeight; // 查询近5分钟的SLO达成率 double sloAttainment metricsStore.getSloAttainment(p, Duration.ofMinutes(5)); double sloDeviation Math.max(0, p.getSloTarget() - sloAttainment); // SLO偏离越大权重上浮越多 double adjustedWeight baseWeight * (1 sloDeviation * AMPLIFICATION_FACTOR); newWeights.put(p, adjustedWeight); } // 归一化保证总和为1 double totalWeight newWeights.values().stream() .mapToDouble(Double::doubleValue).sum(); newWeights.replaceAll((p, w) - w / totalWeight); scheduler.updateWeights(newWeights); } }3.5 反饥饿监控public class StarvationMonitor implements Runnable { private final MapRequestPriority, BlockingQueueScheduledRequest queues; // 记录各优先级最近一次被调度的时间 private final MapRequestPriority, AtomicLong lastDispatchTime; // 饥饿阈值P2超过60秒未被调度视为饥饿 private static final long STARVATION_THRESHOLD_MS 60_000; // 提升持续时间 private static final long BOOST_DURATION_MS 30_000; private volatile RequestPriority boostedPriority null; private volatile long boostExpireTime 0; public StarvationMonitor( MapRequestPriority, BlockingQueueScheduledRequest queues) { this.queues queues; this.lastDispatchTime new EnumMap(RequestPriority.class); for (RequestPriority p : RequestPriority.values()) { lastDispatchTime.put(p, new AtomicLong(System.currentTimeMillis())); } } public void recordDispatch(RequestPriority priority) { lastDispatchTime.get(priority).set(System.currentTimeMillis()); } /** * 检查是否需要饥饿提升 * return 被提升的优先级null表示无需提升 */ public RequestPriority checkStarvation() { long now System.currentTimeMillis(); // 如果正在提升中且未过期保持提升状态 if (boostedPriority ! null now boostExpireTime) { return boostedPriority; } // 重置提升状态 boostedPriority null; // 从低到高检查饥饿 for (RequestPriority p : new RequestPriority[]{ RequestPriority.P2_BATCH, RequestPriority.P1_INTERACTIVE}) { if (!queues.get(p).isEmpty()) { long waitTime now - lastDispatchTime.get(p).get(); if (waitTime STARVATION_THRESHOLD_MS) { boostedPriority p; boostExpireTime now BOOST_DURATION_MS; log.warn(Starvation detected for {}, waitTime{}ms, boosting, p, waitTime); return p; } } } return null; } }四、边界分析与注意事项4.1 权重分配的精度问题概率选择在大样本下符合权重分布但短期内会出现偏差。比如P0权重70%100次调度中可能出现65或75次P0被选中。对严格需要保证配额比例的场景如SLA合同规定的算力分配应改为deficit round-robin算法——每个优先级维护一个赤字计数器被调度一次扣除对应配额确保在任意时间窗口内配额精确可控。4.2 令牌桶的突发 vs 平滑令牌桶的maxBurst参数需要权衡。如果设为速率的2倍如200/s速率400突发允许P0在短时间内以400并发冲击推理引擎可能导致TTFT飙升。我们的建议是maxBurst rate × 1.0即无突发能力严格平滑限流。4.3 权重调整的振荡风险动态权重调整存在潜在的正反馈振荡权重上浮 → P0获得更多GPU → P1被压制 → P1的SLO恶化 → P1权重也上浮 → 权重趋同。防止振荡的关键是设置权重调整的阻尼因子只有SLO偏离超过5%时才调整且每次调整幅度不超过20%。4.4 饥饿提升的优先级反转当P2被饥饿提升到临时高权重时可能导致P1/P0短暂被压制。这是有意为之的权衡——与其让一批请求无限等待不如短暂牺牲SLO。但BOOST_DURATION_MS需要谨慎设置太短10sP2得不到有效服务太长60s会影响P0的SLO。五、总结推理请求调度看似是一个队列问题实则是一个多目标优化问题既要满足SLO、又要防止饥饿、还要在负载波动时自动调整。这套混合方案的三个关键设计令牌桶做入口限流在请求进入调度器之前就做好速率控制避免GPU被单一优先级占满。这是防止饥饿的第一道防线。SLO驱动的动态权重不是静态的70/25/5分配而是根据实时SLO达成率自动浮动的权重。这确保了在负载异常时高优服务体验始终得到保障。反饥饿的Boost机制给低优先级请求一个最长等待时间的保障。在合理的Boost窗口内这是一种可接受的优先级反转。这套调度算法在我们的推理集群上运行了三个月P0的TTFT P99稳定在280ms以内P2的批量任务最长等待时间从没有调度时的无限等待缩短到不超过90秒。文中调度算法已在生产环境验证支撑日均百万级推理请求。令牌桶和优先级队列的实现基于JDK 21的并发工具类。