推理任务优先级队列在线推理、批推理和离线评测要分开排队一、当推理任务挤在同一个队列里谁的体验都保不住调度系统中最隐蔽的坑不是资源不够而是资源分配方式搞错了。在线推理要求 P99 延迟在 200ms 以内批推理允许分钟级等待离线评测甚至可以排队几小时。把这三类任务放进同一个 FIFO 队列批任务会抢占在线推理的 GPU导致延迟飙升反过来在线推理的突发流量也会把离线评测挤到永远排不上队。真实场景的数据足以说明问题某 AI 平台在高峰期同时承载 200 QPS 的在线推理、日均 500 万条数据的批处理、以及持续运行的离线评测流水线。混用单一队列时P99 延迟从 85ms 飙升到 2.3 秒批处理完成时间从 12 分钟膨胀到 47 分钟离线评测任务甚至出现过连续 6 小时无法获得 GPU 的情况。优先级队列不是给重要任务插队这么简单。真正的挑战在于如何定义优先级、如何在优先级之间做公平性保证、以及如何防止低优先级任务永远饥饿。二、多级优先级队列的分层调度模型优先级队列的实现需要三个核心组件分级排队、加权调度、反饥饿机制。调度器使用加权轮询Weighted Round Robin分配 GPU 时间片。默认权重可以是 P0:P1:P2 8:2:1但这只是起点。关键是为每个低优队列引入老化机制当一个任务在队列中等待超过可配置的阈值如 P2 任务等待超过 30 分钟其优先级自动提升一级。同时设置最小保障时间片——每 100 个调度周期中至少给 P2 队列分配 5 个时间片防止极端情况下完全饿死。另一个关键设计是队列容量控制。每个优先级队列都设有最大长度超出时拒绝新任务而不是无限堆积。P0 队列满时直接返回 429 状态码让上游做降级P2 队列满时可以选择丢弃最旧任务或拒绝最新任务取决于业务对数据完整性的要求。三、Go 语言实现的生产级优先级队列调度器以下代码实现了一个基于分层队列 加权调度的推理任务调度器包含完整的反饥饿和过载保护逻辑。package scheduler import ( container/heap context sync time ) // Priority 定义任务优先级 type Priority int const ( PriorityRealtime Priority 0 // 在线推理最高优先 PriorityBatch Priority 1 // 批量推理 PriorityOffline Priority 2 // 离线评测 ) // Task 表示一个推理任务 type Task struct { ID string Priority Priority Payload []byte CreatedAt time.Time Attempts int // 被跳过的次数用于老化 } // TaskQueue 按优先级分层的任务队列 type TaskQueue struct { queues [3]chan *Task maxLen [3]int weights [3]int // 加权轮询权重 starveAt time.Duration // 饥饿阈值 starveAge time.Duration // 老化提升阈值 mu sync.RWMutex stats *QueueStats } type QueueStats struct { mu sync.Mutex Enqueued [3]int64 Dequeued [3]int64 Dropped [3]int64 Starved [3]int64 MaxWait [3]time.Duration } // NewTaskQueue 创建分层优先级队列 func NewTaskQueue(capacity [3]int, weights [3]int) *TaskQueue { q : TaskQueue{ weights: weights, maxLen: capacity, starveAt: 30 * time.Minute, starveAge: 30 * time.Minute, stats: QueueStats{}, } for i : 0; i 3; i { q.queues[i] make(chan *Task, capacity[i]) } return q } // Enqueue 将任务加入对应优先级的队列队列满时拒绝 func (q *TaskQueue) Enqueue(task *Task) error { idx : int(task.Priority) if idx 0 || idx 2 { return ErrInvalidPriority } task.CreatedAt time.Now() select { case q.queues[idx] - task: q.stats.mu.Lock() q.stats.Enqueued[idx] q.stats.mu.Unlock() return nil default: q.stats.mu.Lock() q.stats.Dropped[idx] q.stats.mu.Unlock() return ErrQueueFull } } // Schedule 加权轮询调度器带反饥饿机制 func (q *TaskQueue) Schedule(ctx context.Context) *Task { counter : 0 currentWeight : 0 cycleWeights : q.normalizeWeights() for { // 每 100 个调度周期给低优队列兜底一次 if counter 100 { if task : q.tryDequeueWithStarve(PriorityOffline); task ! nil { return task } counter 0 } // 加权轮询选择当前队列 selected : q.selectByWeight(currentWeight, cycleWeights) task : q.tryDequeueWithStarve(Priority(selected)) if task ! nil { counter currentWeight (currentWeight 1) % len(cycleWeights) return task } // 当前队列为空轮转下一个 currentWeight (currentWeight 1) % len(cycleWeights) if currentWeight 0 { counter } // 全空时短暂等待 select { case -ctx.Done(): return nil case -time.After(10 * time.Millisecond): } } } // normalizeWeights 将权重转为轮询槽位列表 // 例如权重 [8,2,1] 对应 8 个槽位选 P02 个选 P11 个选 P2 func (q *TaskQueue) normalizeWeights() []int { var slots []int for pri, w : range q.weights { for j : 0; j w; j { slots append(slots, pri) } } if len(slots) 0 { return []int{0} // 降级为纯 FIFO } return slots } func (q *TaskQueue) selectByWeight(idx int, slots []int) int { return slots[idx%len(slots)] } // tryDequeueWithStarve 尝试出队同时检查饥饿与老化 func (q *TaskQueue) tryDequeueWithStarve(pri Priority) *Task { idx : int(pri) // 先检查当前队列是否有老化任务需要提升 if pri PriorityBatch { q.promoteStarving(idx) } select { case task : -q.queues[idx]: waitTime : time.Since(task.CreatedAt) q.stats.mu.Lock() q.stats.Dequeued[idx] if waitTime q.stats.MaxWait[idx] { q.stats.MaxWait[idx] waitTime } q.stats.mu.Unlock() return task default: return nil } } // promoteStarving 检查低优队列中是否有饥饿任务并提升 func (q *TaskQueue) promoteStarving(fromIdx int) { if fromIdx 0 { return } select { case task : -q.queues[fromIdx]: if time.Since(task.CreatedAt) q.starveAge { // 老化提升到更高优先级 targetIdx : fromIdx - 1 select { case q.queues[targetIdx] - task: q.stats.mu.Lock() q.stats.Starved[fromIdx] q.stats.mu.Unlock() default: // 目标队列也满了放回原队列 q.queues[fromIdx] - task } } else { // 未达老化阈值放回 q.queues[fromIdx] - task } default: // 队列为空无操作 } } // Worker 从调度器拉取任务并执行 func (q *TaskQueue) Worker(ctx context.Context, workerID int, handler func(*Task) error) { for { task : q.Schedule(ctx) if task nil { return } if err : handler(task); err ! nil { // 生产环境中应接入重试队列或 DLQ continue } } }这个实现的关键权衡通道channel做队列存储天然带背压避免无限制的内存增长老化提升机制保证了长等待任务不会被永久忽略加权轮询提供了可配置的公平性保障。四、优先级划分的边界确认与反模式第一个边界优先级不能无限细分。实践中三类是合理的上限。超过三层后调度开销增长和公平性权衡的复杂性将超过收益。如果某些任务确实需要更细的控制应该在同一个优先级内用子队列 轮询来处理而不是新增优先级层级。第二个边界任务时长差异过大会破坏加权调度的前提假设。如果一个 P0 任务需要 5 秒而另一个 P0 任务需要 0.5 秒简单的加权轮询会让长任务占满时间片。解决方式是在 P0 内部引入最短作业优先SJF策略但前提是你能准确预估任务时长。第三个边界优先级反转。当高优任务依赖于低优任务的结果时低优任务会因为优先级低而排在后面导致高优任务被间接阻塞。这是操作系统调度中的经典问题在推理任务调度中同样存在。解决方案是依赖追踪当检测到 P0 任务等待 P2 任务产出时P2 任务临时继承 P0 的优先级。五、总结推理任务优先级队列不是简单的插队机制而是一个需要同时处理公平性、反饥饿和过载保护的系统工程。三层优先级在线推理、批推理、离线评测配合加权轮询 老化提升是一个经过了充分验证的工程方案。核心需要关注的指标各优先级队列的 P99 延迟、队列最大深度、低优任务饥饿时长。调度器设计的衡量标准不是算法复杂度和智能化程度而是能否在定义的 SLO 范围内稳定运行。