任务调度核心概念：任务、工作流、调度器、执行器
做量化系统，说白了就是跟时间赛跑。你想想看，行情数据每秒都在变，策略信号晚一秒钟，可能就是几百万的差距。所以任务调度这块，是整个中台的「心脏」。
我刚开始搭建量化中台时，对调度这块理解得比较浅。觉得不就是定时跑个脚本嘛，有什么难的？结果第一次上线就出了大问题——某个数据清洗任务卡住了，导致后续所有策略都拿不到最新数据，整整亏了三天。嗯，从那以后，我再也不敢轻视调度系统了。
今天咱们就把四个核心概念掰开揉碎讲清楚：任务、工作流、调度器、执行器。这四个东西，是调度系统的基石。
一、任务（Task）——最小的执行单元
任务是什么？说白了，就是一段需要被执行的逻辑。在量化系统里，任务可以是：
- 拉取某只股票的分钟级行情数据
- 运行一个简单的均线策略
- 计算某个因子的数值
- 发送一条交易信号到下单接口
每个任务都有几个关键属性：
| 属性 | 说明 | 示例 |
|---|---|---|
| 任务ID | 全局唯一标识 | task_20241015_001 |
| 任务类型 | 数据采集/策略计算/信号发送 | DATA_FETCH |
| 执行命令 | 具体要运行的脚本或函数 | python fetch_data.py --symbol=600519 |
| 超时时间 | 任务最长运行时间 | 300秒 |
| 重试策略 | 失败后如何处理 | 最多重试3次，间隔10秒 |
核心要点：任务必须是「原子化」的。什么意思？就是一个任务只做一件事。我曾经见过有人把数据清洗、策略计算、下单全部写在一个任务里，结果出了问题根本不知道是哪一步挂了。拆开，每个任务只干一件事，出了问题定位快得多。
代码层面，一个任务通常长这样：
class Task:
def __init__(self, task_id, task_type, command, timeout=300):
self.task_id = task_id
self.task_type = task_type
self.command = command
self.timeout = timeout
self.status = 'pending' # pending, running, success, failed
self.retry_count = 0
self.max_retries = 3
def execute(self):
# 实际执行逻辑
pass二、工作流（Workflow）——任务的编排艺术
单个任务能做的事有限。真正的量化系统，需要多个任务协同工作。工作流就是把这些任务串起来，定义它们的执行顺序和依赖关系。
举个例子，一个典型的日频策略工作流：
- 任务A：拉取全市场股票行情数据（9:00执行）
- 任务B：清洗数据，剔除停牌股（A完成后执行）
- 任务C：计算50个技术因子（B完成后执行）
- 任务D：运行选股模型，生成信号（C完成后执行）
- 任务E：发送交易指令到券商（D完成后，14:55执行）
你看，这里就有明显的依赖关系。B依赖A，C依赖B，以此类推。如果A失败了，B、C、D、E都没必要跑了。
我的经验：工作流设计时，一定要考虑「失败传播」的问题。我习惯在每个工作流里加一个「兜底任务」——如果主流程失败，就触发一个告警任务，发短信或者钉钉通知值班人员。别问我为什么，问就是吃过亏。
工作流的核心数据结构：
class Workflow:
def __init__(self, workflow_id, name):
self.workflow_id = workflow_id
self.name = name
self.tasks = [] # 任务列表
self.dag = {} # 依赖关系图，key是任务ID，value是依赖列表
def add_task(self, task, depends_on=None):
self.tasks.append(task)
self.dag[task.task_id] = depends_on or []
def get_execution_order(self):
# 拓扑排序，确定执行顺序
pass三、调度器（Scheduler）——整个系统的大脑
调度器负责回答三个问题：
- 什么时候执行？——定时触发还是事件触发
- 谁先执行？——优先级和依赖关系
- 失败了怎么办？——重试、跳过还是告警
调度器的工作流程大致是这样的：
# 伪代码，展示调度器核心逻辑
class Scheduler:
def __init__(self):
self.workflows = {}
self.running_tasks = set()
def add_workflow(self, workflow, cron_expr):
# cron_expr: "0 9 * * 1-5" 表示周一到周五9点执行
self.workflows[workflow.workflow_id] = {
'workflow': workflow,
'cron': cron_expr
}
def tick(self):
# 每秒检查一次，看哪些工作流需要触发
now = datetime.now()
for wf_id, config in self.workflows.items():
if self.should_trigger(config['cron'], now):
self.execute_workflow(config['workflow'])
def execute_workflow(self, workflow):
# 按拓扑排序执行任务
order = workflow.get_execution_order()
for task_id in order:
task = workflow.get_task(task_id)
self.execute_task(task)
def execute_task(self, task):
# 把任务交给执行器
executor = self.get_executor(task.task_type)
executor.run(task)注意：调度器本身不能成为单点故障。我见过最惨的一次，调度器进程挂了，所有定时任务都没跑，直到第二天才发现。所以调度器一定要做高可用，至少主备切换。
四、执行器（Executor）——干活的工人
调度器只负责「安排」，真正干活的是执行器。执行器拿到任务后，负责：
- 分配资源（CPU、内存、网络）
- 运行任务脚本
- 监控运行状态
- 收集执行结果
- 上报日志
在量化系统里，我通常把执行器分成几种类型：
| 执行器类型 | 适用场景 | 资源消耗 |
|---|---|---|
| 本地执行器 | 轻量任务，如数据校验 | 低 |
| 进程执行器 | 策略计算，需要隔离 | 中 |
| 容器执行器 | 复杂任务，需要环境隔离 | 高 |
| 分布式执行器 | 大规模并行计算 | 极高 |
执行器的核心逻辑：
class Executor:
def __init__(self, executor_type='local'):
self.type = executor_type
self.max_concurrency = 5 # 最大并行数
self.running = {}
def run(self, task):
if len(self.running) >= self.max_concurrency:
# 队列满了，等待
return self.queue(task)
# 创建子进程/线程执行
process = self.create_process(task)
self.running[task.task_id] = process
# 监控执行状态
process.start()
process.join(timeout=task.timeout)
if process.is_alive():
# 超时了，强制杀掉
process.terminate()
task.status = 'timeout'
else:
task.status = 'success' if process.exitcode == 0 else 'failed'
del self.running[task.task_id]
return task.status避坑指南：我曾经犯过一个错误——所有任务都用同一个执行器。结果一个耗内存的策略把整个系统搞崩了。后来我改成按任务类型分配不同的执行器池，数据采集用轻量池，策略计算用重型池，互相隔离，再也没出过类似问题。
五、四个概念的关系
这四个概念的关系，我用一张图来展示：
从图中可以看得很清楚：任务是最小单元，工作流把任务串起来，调度器决定什么时候跑，执行器负责实际干活。四个角色各司其职，缺一不可。
我的建议：刚开始做调度系统时，别想着一步到位。先实现最简单的「定时任务+本地执行」，跑通了再慢慢加工作流、分布式执行器。我见过太多人一上来就搞Kubernetes集群，结果光环境搭建就花了两周，业务逻辑一行没写。先跑起来，再优化，这是做量化系统的铁律。
好了，这四个核心概念就讲到这里。记住它们的职责和关系，后面讲具体实现时，你会觉得豁然开朗。
公众号:蓝海资料掘金营，微信deep3321