推理请求调度策略按用户优先级还是按预估耗时一、所有请求看起来一样但他们的等待成本完全不同一个标准的推理请求系统面临两难选择VIP 用户的请求应该优先处理还是预估耗时短的请求应该优先处理前者是优先级策略保证高价值用户的体验后者是最短作业优先策略最大化系统吞吐量。两者的冲突在实际场景中非常尖锐。一个付费企业用户的复杂推理请求预估 800ms和一个免费用户的简单请求50ms同时到达。按优先级策略复杂请求立即执行简单请求等待 800ms 后处理——吞吐量被牺牲了。按最短作业优先简单请求 50ms 完成复杂请求 50ms 后开始——VIP 用户多等了 50ms。极端情况下如果按最短作业优先且系统持续有短任务涌入长任务可能永远得不到执行——这就是经典的任务饥饿。反过来如果只按用户优先级调度高峰期系统的平均延迟会显著上升因为排队模型从短任务快速通过变成了高优任务插队其他任务排长队。二、多级反馈队列的混合调度策略将操作系统中经过数十年验证的多级反馈队列MLFQ思想引入推理请求调度在吞吐和公平之间取得平衡。多级反馈队列的核心思想请求的优先级不是静态的它在队列中的等待时间越长优先级就动态提升。VIP 用户的请求走快速通道Q0但不会完全阻塞普通用户——普通用户的短任务走 Q2 快速通道长任务走 Q4 但有老化机制保证不会无限等待。调度器在每个调度周期做两件事一、按队列权重Q0:Q2:Q3:Q4 5:3:1:1选择下一个要执行的任务二、扫描 Q3 和 Q4将等待超过阈值的任务提升至上一级队列。三、Go 语言实现的多级反馈队列推理调度器package scheduler import ( container/heap context sync time ) // RequestPriority 请求优先级 type RequestPriority int const ( PriVIP RequestPriority 0 PriNormal RequestPriority 1 ) // InferenceRequest 推理请求 type InferenceRequest struct { ID string UserID string UserPriority RequestPriority EstimatedMS int64 // 预估执行耗时毫秒 Payload []byte ArrivalTime time.Time Promotions int // 被老化提升的次数 QueueIndex int // 当前所在队列 } // RequestHeap 按等待时间排序的优先队列 type RequestHeap []*InferenceRequest func (h RequestHeap) Len() int { return len(h) } func (h RequestHeap) Less(i, j int) bool { return h[i].ArrivalTime.Before(h[j].ArrivalTime) } func (h RequestHeap) Swap(i, j int) { h[i], h[j] h[j], h[i] } func (h *RequestHeap) Push(x interface{}) { *h append(*h, x.(*InferenceRequest)) } func (h *RequestHeap) Pop() interface{} { old : *h n : len(old) x : old[n-1] *h old[:n-1] return x } // MLFQScheduler 多级反馈队列调度器 type MLFQScheduler struct { mu sync.Mutex // 5级队列Q0 VIP快速, Q1 未用预留, Q2 短任务, Q3 中任务, Q4 长任务 queues [5]*RequestHeap weights [5]int starveAge time.Duration // 老化阈值 // 队列容量 maxLen [5]int // 统计 stats *MLFQStats } type MLFQStats struct { mu sync.Mutex TotalProcessed int64 AvgWaitMS int64 Promotions int64 VIPLatency []time.Duration NormalLatency []time.Duration } // NewMLFQScheduler 创建调度器 func NewMLFQScheduler() *MLFQScheduler { s : MLFQScheduler{ weights: [5]int{5, 0, 3, 1, 1}, starveAge: 30 * time.Second, maxLen: [5]int{1000, 0, 5000, 2000, 1000}, stats: MLFQStats{}, } for i : range s.queues { h : RequestHeap{} heap.Init(h) s.queues[i] h } return s } // Enqueue 将请求放入合适的队列 func (s *MLFQScheduler) Enqueue(req *InferenceRequest) error { s.mu.Lock() defer s.mu.Unlock() req.ArrivalTime time.Now() queueIdx : s.assignQueue(req) req.QueueIndex queueIdx h : s.queues[queueIdx] if h.Len() s.maxLen[queueIdx] { return ErrQueueFull } heap.Push(h, req) return nil } // assignQueue 根据用户优先级和预估耗时分配队列 func (s *MLFQScheduler) assignQueue(req *InferenceRequest) int { // VIP 用户直接进入 Q0 快速通道 if req.UserPriority PriVIP { return 0 } // 普通用户按预估耗时分级 switch { case req.EstimatedMS 100: return 2 // Q2: 短任务优先 case req.EstimatedMS 1000: return 3 // Q3: 中任务 default: return 4 // Q4: 长任务 } } // Next 选取下一个要执行的任务 func (s *MLFQScheduler) Next(ctx context.Context) *InferenceRequest { s.mu.Lock() defer s.mu.Unlock() // 第1步老化扫描 — 将等待过久的任务提升 s.scanAndPromote() // 第2步加权遍历各队列选择任务 for _, qIdx : range s.weightedOrder() { h : s.queues[qIdx] if h.Len() 0 { continue } req : heap.Pop(h).(*InferenceRequest) // 记录等待时间用于 SLO 监控 waitTime : time.Since(req.ArrivalTime) s.stats.mu.Lock() s.stats.TotalProcessed s.stats.AvgWaitMS (s.stats.AvgWaitMS waitTime.Milliseconds()) / 2 if req.UserPriority PriVIP { s.stats.VIPLatency append(s.stats.VIPLatency, waitTime) } else { s.stats.NormalLatency append(s.stats.NormalLatency, waitTime) } s.stats.mu.Unlock() return req } return nil } // weightedOrder 按权重生成队列遍历顺序权重高的更可能被先选中 func (s *MLFQScheduler) weightedOrder() []int { // 简化实现权重越大的队列排在越前面 order : []int{0, 1, 2, 3, 4} // 按权重降序排列 for i : 0; i len(order); i { for j : i 1; j len(order); j { if s.weights[order[i]] s.weights[order[j]] { order[i], order[j] order[j], order[i] } } } return order } // scanAndPromote 扫描低优队列将等待过久的任务提升 func (s *MLFQScheduler) scanAndPromote() { now : time.Now() // 检查 Q4 (长任务队列) h4 : s.queues[4] var promoted []*InferenceRequest var kept RequestHeap for h4.Len() 0 { req : heap.Pop(h4).(*InferenceRequest) if now.Sub(req.ArrivalTime) s.starveAge { // 等待超过阈值提升到 Q3 req.Promotions req.QueueIndex 3 promoted append(promoted, req) s.stats.mu.Lock() s.stats.Promotions s.stats.mu.Unlock() } else { kept append(kept, req) } } // Q4 中未被提升的放回去 heap.Init(kept) s.queues[4] kept // 提升的任务加入 Q3 h3 : s.queues[3] for _, req : range promoted { heap.Push(h3, req) } // 同样检查 Q3 中是否有等待过久的提升到 Q2 h3 s.queues[3] promoted nil kept nil for h3.Len() 0 { req : heap.Pop(h3).(*InferenceRequest) if now.Sub(req.ArrivalTime) s.starveAge*2 { // Q3 中的任务等待超过 2 倍阈值提升到 Q2 req.Promotions req.QueueIndex 2 promoted append(promoted, req) } else { kept append(kept, req) } } heap.Init(kept) s.queues[3] kept h2 : s.queues[2] for _, req : range promoted { heap.Push(h2, req) } } // Worker 从调度器拉任务并执行 func (s *MLFQScheduler) Worker(ctx context.Context, handler func(*InferenceRequest) error) { for { select { case -ctx.Done(): return default: } req : s.Next(ctx) if req nil { // 所有队列为空短暂休眠 select { case -ctx.Done(): return case -time.After(10 * time.Millisecond): } continue } if err : handler(req); err ! nil { // 失败处理可重试或放入死信队列 continue } } }四、预估耗时的准确性与调度效果多级反馈队列的核心假设是你能准确预估任务耗时。预估偏差会导致调度效果下降高估长任务一个实际 50ms 的任务被错误分类到 Q4导致不必要的长等待低估短任务一个实际 800ms 的任务进入 Q2短任务队列占用了本应服务真正短任务的时间片预估耗时的数据来源历史数据同用户同类请求的历史平均耗时、模型特征输入 Token 数 * 单 Token 推理耗时、系统反馈首次请求后的实际耗时反馈给调度器更新预估。建议使用指数加权移动平均EWMA持续校准预估模型。另一个需要考虑的点是请求可以中途抢占。如果一个 Q2 中的任务执行超过 100ms 仍未完成应该被抢占并降级到 Q3让真正的短任务继续快速通过。这要求 Worker 支持检查点机制或至少支持超时中断。五、总结推理请求调度不应在用户优先级和预估耗时之间二选一。多级反馈队列提供了兼顾两者的实用方案VIP 用户的请求走快速通道保证 SLA普通用户的短任务快速通过保证吞吐长任务通过老化机制防止饥饿。预估耗时的准确性是这套方案效果的核心变量——误差越大调度质量越差。实际部署中应持续监控各队列的等待时间分布动态调整权重和队列划分阈值。