推理并发调度系统化实践 - 第3节
3. 任务模型与状态管理
任务模型，说白了就是给任务画一张「人生轨迹图」。
我刚开始做分布式调度时，犯过一个低级错误——把任务状态定义得太简单。就仨状态：等待、运行、完成。结果呢？任务挂了怎么办？网络超时怎么办？节点宕机怎么办？全乱套了。
一个成熟的任务生命周期，至少包含这些阶段：
- 创建（Created）：任务刚被提交，还没做任何处理
- 等待（Pending）：已通过校验，排队等待调度
- 就绪（Ready）：所有前置依赖已满足，可以执行了
- 运行（Running）：正在被某个工作节点执行
- 完成（Completed）：正常执行完毕
- 失败（Failed）：执行过程中出错
- 取消（Cancelled）：被用户或系统主动终止
- 超时（TimedOut）：超过最大执行时间
你想想看，如果只有「成功」和「失败」两种结局，那任务被取消算啥？超时又算啥？这些边界情况，恰恰是系统健壮性的试金石。
核心原则：任务状态必须覆盖所有可能的终态，不能有「未定义」的灰色地带。
任务状态机设计
状态机，就是给任务状态转换画一张「交通图」。哪些路能走，哪些路是死胡同，必须清清楚楚。
我习惯用有限状态机（FSM）来建模。每个任务实例内部维护一个状态变量，只有合法的转换才能被执行。
状态转换规则
这张表是我在项目中反复打磨出来的：
| 当前状态 | 允许的下一个状态 | 触发条件 |
|---|---|---|
| Created | Pending | 任务校验通过 |
| Pending | Ready, Cancelled | 依赖满足 / 用户取消 |
| Ready | Running, Cancelled | 调度器分配 / 用户取消 |
| Running | Completed, Failed, TimedOut | 正常结束 / 异常 / 超时 |
| Completed | （终态） | — |
| Failed | Pending（重试） | 重试策略允许 |
| Cancelled | （终态） | — |
| TimedOut | Pending（重试） | 重试策略允许 |
我曾经踩过的坑：允许 Failed 状态直接跳转到 Running。结果任务重试时，旧的工作节点还在跑，新节点又启动了，两个进程写同一个输出文件，数据全乱了。记住：重试必须先回到 Pending，重新排队。
状态机代码骨架
嗯，这里给一个简单的状态机实现思路：
class TaskStateMachine:
def __init__(self):
self.state = TaskState.CREATED
self.transitions = {
TaskState.CREATED: [TaskState.PENDING],
TaskState.PENDING: [TaskState.READY, TaskState.CANCELLED],
TaskState.READY: [TaskState.RUNNING, TaskState.CANCELLED],
TaskState.RUNNING: [TaskState.COMPLETED, TaskState.FAILED, TaskState.TIMED_OUT],
TaskState.FAILED: [TaskState.PENDING], # 仅允许重试
TaskState.TIMED_OUT:[TaskState.PENDING], # 仅允许重试
}
def transition_to(self, new_state):
if new_state not in self.transitions[self.state]:
raise IllegalStateTransition(
f"不能从 {self.state} 转到 {new_state}"
)
self.state = new_state为什么要把转换规则硬编码？因为状态机最怕的就是「意外跳转」。你想想看，如果因为一个 bug，任务从 Completed 跳到了 Running，那下游系统会收到两次完成通知，后果不堪设想。
任务优先级模型
优先级，说白了就是「谁先谁后」的问题。但实际项目中，优先级远不止一个数字那么简单。
多级优先级队列
我建议用「优先级 + 时间戳」的组合策略。为什么呢？
- 纯优先级调度：高优先级任务永远插队，低优先级任务可能饿死
- 纯 FIFO：无法应对紧急任务
- 优先级 + 时间戳：同优先级内按时间排序，不同优先级按等级排序
我在项目中遇到过这样一个场景：一个低优先级的批量任务，跑了 10 个小时，结果被源源不断的高优先级查询任务反复打断，最后跑了 3 天都没跑完。这就是典型的「饥饿问题」。
解决方案：引入「优先级老化」机制。任务等待时间越长，优先级自动提升。这样既保证了紧急任务优先，又不会让低优先级任务永远得不到执行。
优先级继承
还有一个容易被忽略的点——优先级继承。假设任务 A（高优先级）依赖任务 B（低优先级），如果 B 一直排在后边，A 就得干等着。这时候，B 应该临时继承 A 的优先级，尽快执行完。
嗯，这个机制在实时系统中尤其重要。我见过一个视频处理流水线，因为没做优先级继承，一个高优先级的直播任务被低优先级的转码任务堵了 30 秒，直播直接卡断。
任务依赖关系管理
依赖关系，是任务调度中最复杂的部分。没有之一。
依赖类型
常见的依赖关系有这几种：
- 顺序依赖：B 必须在 A 之后执行（A → B）
- 并行依赖：B 和 C 可以同时执行，但都必须在 A 之后
- 条件依赖：B 只在 A 成功时才执行，否则执行 C
- 资源依赖：B 需要 A 释放某个资源才能执行
DAG 调度
大多数任务依赖关系可以用有向无环图（DAG）来表示。每个节点是一个任务，每条边是一个依赖关系。
下面是我手绘的一张 DAG 调度流程图：
这张图展示了一个典型的 DAG 调度场景：
- 任务 A 是根节点，没有依赖
- 任务 B 和 C 都依赖 A，它们可以并行执行
- 任务 D 依赖 B 和 C，必须等两者都完成
- 任务 E 依赖 D，是最后一个任务
依赖解析算法
我常用的依赖解析算法是「拓扑排序 + 入度计数」：
def resolve_dependencies(tasks):
# 构建入度表
in_degree = {task.id: 0 for task in tasks}
adjacency = {task.id: [] for task in tasks}
for task in tasks:
for dep_id in task.dependencies:
adjacency[dep_id].append(task.id)
in_degree[task.id] += 1
# 找到所有入度为0的任务（没有依赖）
ready_queue = [tid for tid, deg in in_degree.items() if deg == 0]
while ready_queue:
tid = ready_queue.pop(0)
# 调度执行 task tid
for next_tid in adjacency[tid]:
in_degree[next_tid] -= 1
if in_degree[next_tid] == 0:
ready_queue.append(next_tid)我曾经踩过的坑：没有处理循环依赖。结果调度器死循环了，CPU 飙到 100%，整个集群都卡住了。后来我加了一个「最大迭代次数」的保护，超过阈值直接报错。记住：DAG 必须是无环的，有环就是 bug。
依赖超时与降级
实际项目中，依赖的任务可能永远不结束。比如上游任务挂了，或者网络分区了。这时候怎么办？
我建议引入「依赖超时」机制：
- 给每个依赖关系设置一个最大等待时间
- 超时后，可以选择「跳过依赖」或「执行默认值」
- 记录超时事件，方便事后排查
举个例子：一个数据报表任务，依赖上游的数据采集任务。如果采集任务超时了，报表任务可以先用昨天的数据生成报表，等数据到了再更新。这叫「优雅降级」。
核心思想：依赖管理不是「非黑即白」的。有时候，部分成功比完全失败更有价值。
好了，任务模型和状态管理这块，说白了就是给任务画好「人生轨迹」，定好「交通规则」，排好「优先级」，理清「人际关系」。把这些基础打牢了，上层调度逻辑才能跑得稳。