分布式调度器设计

type Task struct {
    ID            string            `json:"id"`
    Queue         string            `json:"queue"`          // 所属队列
    Priority      int               `json:"priority"`       // 数值越大优先级越高
    CPUReq        int64             `json:"cpuReq"`         // 所需 CPU 毫核
    MemReq        int64             `json:"memReq"`         // 所需内存 MB
    Dependencies  []string          `json:"dependencies"`   // 依赖任务 ID
    Deadline      int64             `json:"deadline"`       // 截止时间戳
    Affinity      map[string]string `json:"affinity"`       // 节点亲和（标签）
    MaxRetries    int               `json:"maxRetries"`
    Timeout       int64             `json:"timeout"`        // 单次执行超时
    Payload       string            `json:"payload"`
    State         TaskState         `json:"state"`
}

func (d *DAG) ReadyTasks() []string {
    ready := []string{}
    for id, node := range d.Nodes {
        if node.InDegree == 0 && !node.Done && node.Running == 0 {
            ready = append(ready, id)
        }
    }
    return ready
}

public class WeightedFairQueue {
    // 每个队列维护一个虚拟运行时间，调度时选虚拟时间最小的队列出队
    private final Map<String, QueueState> queues = new ConcurrentHashMap<>();

    public synchronized Task scheduleNext() {
        String winner = null;
        long minVirtual = Long.MAX_VALUE;
        for (var e : queues.entrySet()) {
            if (!e.getValue().isEmpty()) {
                long v = e.getValue().virtualTime / e.getValue().weight;
                if (v < minVirtual) {
                    minVirtual = v;
                    winner = e.getKey();
                }
            }
        }
        if (winner == null) return null;
        Task t = queues.get(winner).dequeue();
        queues.get(winner).virtualTime += 100; // 每出队一个任务累加
        return t;
    }
}

func (s *Scheduler) preempt(high *Task) bool {
    victims := s.findVictims(high)      // 找低优 RUNNING 任务
    sort.Slice(victims, func(i, j int) bool {
        if victims[i].Priority != victims[j].Priority {
            return victims[i].Priority < victims[j].Priority
        }
        // 快完成的任务优先保留：剩余运行时长排序
        return victims[i].RemainingEst > victims[j].RemainingEst
    })
    freed := int64(0)
    for _, v := range victims {
        s.kill(v, "preempted by "+high.ID)   // 发起优雅退出
        freed += v.CPUReq + v.MemReq
        if freed >= high.CPUReq+high.MemReq {
            return true
        }
    }
    return false
}

// 最小堆实现延迟队列
public class DelayQueue {
    private final PriorityQueue<DelayedTask> heap = new PriorityQueue<>(
        Comparator.comparingLong(DelayedTask::getDeadline));

    public void add(DelayedTask task) {
        heap.offer(task);
    }

    public List<DelayedTask> pollExpired(long now) {
        List<DelayedTask> ready = new ArrayList<>();
        while (!heap.isEmpty() && heap.peek().getDeadline() <= now) {
            ready.add(heap.poll());
        }
        return ready;
    }
}

func nextRetryDelay(attempt int, maxBackoff time.Duration) time.Duration {
    exp := time.Duration(1<<uint(min(attempt, 10))) * time.Second
    if exp > maxBackoff {
        exp = maxBackoff
    }
    // 加抖动，避免重试风暴
    jitter := time.Duration(rand.Int63n(int64(exp / 4)))
    return exp/2 + jitter
}

func (s *Scheduler) OnNodeDown(nodeID string) {
    for _, t := range s.runningOnNode(nodeID) {
        // 依据任务类型决定：可重跑 → 重新入队；不可重跑 → 标记 FAILED 告警
        if t.Restartable {
            s.enqueue(t, WithRetryReason("node down: "+nodeID))
        } else {
            s.markFailed(t, "node down, non-restartable")
        }
    }
    // 更新资源视图，把节点标记不可用
    s.nodes.MarkDown(nodeID)
}

type Scheduler struct {
    queue   *WeightedFairQueue
    nodes   *ResourceManager
    store   *TaskStore
    delay   *TimeWheel
}

func (s *Scheduler) Run(ctx context.Context) {
    for {
        select {
        case <-ctx.Done():
            return
        case <-time.After(50 * time.Millisecond): // 调度节拍
            s.tick()
        }
    }
}

func (s *Scheduler) tick() {
    now := time.Now()
    // 1) 时间轮到期任务进入就绪队列
    for _, t := range s.delay.PollExpired(now.UnixMilli()) {
        s.queue.Enqueue(t)
    }
    // 2) 尝试调度就绪任务
    for {
        task := s.queue.Next()
        if task == nil {
            break
        }
        node := s.nodes.BestFit(task)
        if node == nil {
            // 资源不足：尝试抢占高优前的低优任务
            if task.Priority >= PreemptThreshold && s.preempt(task) {
                continue
            }
            s.queue.RequeueBack(task) // 放回队首等待
            break
        }
        s.dispatch(task, node)
    }
}