如果你也在维护一堆定时任务散落在 crontab 里、脚本目录里、甚至某些只能手动点一下的页面上每天最怕的就是某个任务没跑、跑重了、或者跑挂了没人知道。这期分享的项目代号是 Apollo-11是我花了大半年折腾出来的一个轻量级任务编排与数据归档工具。它解决的问题很具体把不同类型的批处理任务统一成一个可配置、可重试、可观测的流水线用完它之后我这边基本告别了靠盯日志救命的阶段。Apollo-11 这个名字没什么复杂的含义纯属内部按希腊神话太阳神加版本号随手拼的听着顺口就沿用下来了。它最适合两类人一是手里有几十上百个脚本任务、想找个轻量方案统一管起来的人二是刚开始接触任务调度、不想一上来就上重型分布式框架的人。如果你面对的是上百个节点、每天几十万任务的规模那可能得去看更专业的调度系统但如果你和我一样单机或几台机器就够用这个设计和实现思路应该能给你不少参考。整套系统用 Python 写的存储层只用了内置的 SQLite没有引入额外中间件。好处是部署极简单一个进程就能跑起来坏处是会有一些奇奇怪怪的锁问题但我在第 5 章里会专门讲怎么踩过去。下面我从为什么要做这个项目开始把架构、实现、部署和踩坑都过一遍。1. 为什么会有 Apollo-11 这个项目1.1 从散落的脚本到无法收场的定时任务最早我手上的任务其实不算多大概二十来个。有的任务是每天凌晨拉数据有的是每小时做一次归档还有几个是外部接口回调后需要立即触发。它们各自独立散落在不同服务器的 crontab 里每个脚本又用自己的方式写日志。刚开始还好一旦某个任务开始报错真正麻烦的是排查链路得先翻 crontab 看到底有没有配再看脚本日志再去看系统时间有时候半天过去才发现根本原因是两个脚本同时跑把同一个文件锁冲突了。这种状态最难受的不是单次失败而是没有全局视角。你不知道今天总共有多少任务该跑跑了几个成功几个失败几个重试了几次全部要靠人肉去对。于是我开始琢磨能不能做一个很小的统一入口把所有脚本都变成一个个“任务”由一个中心调度器来安排运行顺序、记录状态、控制重试。1.2 立项时的目标与边界我给自己定了几个很克制的目标。第一配置化新加一个任务尽量只写一个配置文件不改调度器代码。第二插件化每个任务的真正逻辑放在独立模块里通过约定方式注册进来。第三可观测任务从创建到成功或失败的所有状态变化都要落库方便事后复盘。第四能在单机低资源环境下稳定跑不依赖外部数据库、消息队列和容器编排。同时我也明确“不做什么”。不做分布式调度不做实时流处理不做 Web 前端。当时只需要把批处理这一条链路做扎实前端界面以后需要再说。很多人一上来就想搞高可用、故障转移、分布式锁实际上如果你的任务规模只有几十到几千单机加一个好设计完全够用架构简单本身就是最大的可靠性来源。2. 整体设计与关键决策2.1 三个模块的职责划分Apollo-11 从结构上拆成三块任务定义层、调度执行层、存储观测层。任务定义层负责解析 YAML 配置、加载插件、校验参数调度执行层负责从队列里取任务、控制并发、处理超时和重试存储观测层负责状态持久化、日志落库、心跳记录。三层之间只通过预先定义好的数据结构和数据库表交互不互相调用内部方法。模块主要职责对外接口任务定义层解析配置、加载插件、参数校验配置目录 Python 注册函数调度执行层领任务、分发、并发控制、超时回收调度器主循环 Worker 进程存储观测层状态持久化、日志、心跳、审计SQLite 表 查询接口这样切的原因很实际我可以不开 Web 界面直接查数据库表就能看到一切想让某个任务不跑改配置加一个 enabled: false 就行想加新任务类型不会碰调度器核心逻辑。对于一个小项目来说这种边界足够清晰又不会产生过度设计。2.2 为什么用状态机管理任务生命周期任务不能只存一个“成功还是失败”的标签。在真实运行里一个任务会经历等待、运行中、等待重试、成功、失败、被取消等状态而且状态的变迁是有方向的。比如一个任务不会从成功回到运行中但可以从失败进入等待重试再从等待重试进入运行中。我直接用状态机来约束把可能的状态和合法迁移写死避免脏数据。任务表里的 status 字段用字符串表示值包括 pending、running、retry_wait、success、failed、canceled、dead。调度器只允许按状态机定义的方向更新任何非法更新直接报错。这样做的最大好处是如果调度进程中途挂了重启后可以扫描 running 状态的任务发现心跳已经过期就把状态改回 pending 或 retry_wait重新执行而不是永远卡死在那。状态机也天然适合做幂等。每个任务执行前会生成一个执行批次号execution_id所有插件的副作用操作都要带上这个批次号。即使同一个任务因为异常被调度了两次执行记录也能通过批次号区分开不会把两次运行结果混在一起。2.3 插件化执行器的实现思路执行器本质上就是一个可调用对象输入是任务参数输出是结果或者抛异常。为了新增任务不动调度器我用了一个很朴素的注册表模式插件模块放在 tasks/ 目录下每个模块里定义一个 register 函数调度器在启动时扫描目录把所有插件登记到一个 dict 里key 是任务类型名。比如你有一个任务类型叫 archive_files插件模块里就定义一个同名处理器。YAML 配置里写 type: archive_files调度器拿到后去注册表查对应的处理函数然后带着参数执行。这里不需要复杂的插件框架Python 的 importlib 加约定式目录扫描就够了。如果你以后要支持更多的执行方式也可以把“处理函数”抽象成“执行器类”但现阶段函数已经够用。3. 核心功能实现与实操细节3.1 任务配置用 YAML 描述一个可调度任务在 Apollo-11 里每个任务对应一个 YAML 文件放在 jobs 目录下。一个典型的归档任务长这样id: nightly_backup type: archive_files enabled: true schedule: 0 3 * * * priority: 3 timeout: 1200 retry: max_attempts: 3 delay_seconds: 60 params: source_dir: /data/raw target_dir: /data/archive dedup_key: sha256字段数量不多但每条都经过实际运行打磨。id 是全局唯一标识type 对应插件注册名schedule 用 cron 表达式表达触发规则priority 是数字越小越先执行timeout 是单个任务的最大运行时长超过就会被强制结束retry 控制失败后的重试次数和初始等待时间params 是传给插件函数的自由字典。这里有一个细节值得展开schedule 字段并不是调度器主动依赖的。我实现了一个轻量级 cron 解析器调度器每分钟醒来一次检查哪些任务到了触发时间把它们的下一次运行时间写入队列。这样即使调度器重启只要数据库里存有上次计算出的 next_run_at也不会漏掉本该执行的任务。触发时间计算是所有定时系统最容易出错的地方我建议不管用什么方案都要把 next_run_at 持久化下来而不是每次临时算。3.2 调度循环与并发控制调度器主循环是一个简单的 while True每 5 秒扫描一次任务队列把到期且未被领取的任务标记为 running然后交给 Worker 进程池执行。为了避免多进程同时修改 SQLite 导致锁冲突我做了两个关键设置一是数据库开启 WAL 模式让读写并发更宽松二是设置 busy_timeout让写入碰到锁时等待而不是立刻报错。取任务时不能简单地 SELECT 出 pending 状态的任务然后 UPDATE因为两个进程可能同时看到同一条记录导致重复执行。我采用了一个更保守的做法先把任务状态 UPDATE 成 reserved如果影响行数是 1说明这个任务被我抢到了如果影响行数是 0说明已经被别人抢走就跳过。这个“先占坑再干活”的思路在关系型数据库里非常通用。# 核心领取函数简化版 def claim_task(conn, worker_id, now): conn.execute(BEGIN IMMEDIATE) row conn.execute( SELECT id FROM tasks WHERE status pending AND next_run_at ? AND enabled 1 ORDER BY priority ASC, next_run_at ASC LIMIT 1 , (now,), ).fetchone() if row is None: conn.rollback() return None conn.execute( UPDATE tasks SET status reserved, claimed_by ?, heartbeat_at ? WHERE id ?, (worker_id, now, row[id]), ) conn.commit() return row[id]Worker 拿到任务 ID 后会先通过一个内部队列再领取一次这里其实还有一层“内存队列”用来平滑数据库的读写压力。Worker 进程在执行期间每 10 秒更新一次心跳时间。如果 worker 崩溃调度器在扫描过期心跳时就会把任务重新放回 pending可以让另一个 worker 接管。这里要特别注意的是任务本身必须支持被重新执行也就是要有幂等设计否则“恢复”可能造成比崩溃更严重的问题。3.3 重试、幂等与告警任务执行失败后的等待时间我采用了指数退避加随机抖动避免多个失败任务同时重试造成资源尖峰。第一次重试延迟是配置里的 delay_seconds后面每次翻倍并加上 0 到 10 秒的随机偏移。这样比固定间隔更稳实际跑下来失败任务对整体系统的影响小很多。def next_retry_delay(attempt: int, base_delay: int) - int: import random return base_delay * (2 ** (attempt - 1)) random.randint(0, 10)幂等设计分两层。第一层是任务级幂等任务 ID 加 execution_id 组成全链路唯一键插件执行前先把这个键写入执行记录表拿到返回值后更新对应记录。第二层是业务级幂等比如归档任务里有一个 dedup_key同一个源文件不会重复归档到目标目录而是先比对指纹文件如果已经存在就直接跳过。如果你做的任务不是归档而是发通知、写外部接口那更要谨慎尽量让外部接口也支持用 execution_id 做去重。告警是必须的不然任务失败了还是没人知道。最简单的实现是自定义 Webhook任务连续失败两次后调度器把任务 ID、错误信息、执行日志摘要组装成一个 JSON POST 到配置好的 Webhook 地址再配合常见的群机器人或通知服务就能把消息推到你手机上。不要太早在这个模块上堆功能先保证“失败必达”等实际用了再考虑告警分级别、告警合并这些。4. 部署运行与稳定性表现4.1 最小部署结构与启动方式Apollo-11 的部署结构非常简单一台 Linux 服务器Python 3.10 以上装好依赖后建目录 jobs/ 和 tasks/然后启动两个进程一个是 scheduler一个是 worker 池。scheduler 负责扫描队列、触发定时任务、管理状态机worker 进程池负责真正执行任务可以随时手动扩缩。目录结构长这样apollo11/ ├── config.yaml ├── jobs/ # 放任务 YAML ├── tasks/ # 放插件模块 ├── logs/ # 运行日志 ├── data/ │ └── apollo11.db # SQLite 数据文件 └── apollo11/ ├── scheduler.py ├── worker.py └── plugin_loader.py启动命令也很直接执行python -m apollo11.scheduler启动调度器执行python -m apollo11.worker --processes 4启动四个 worker 进程。如果你希望开机自启用系统自带的服务管理工具配一个服务单元就行不复杂。这里唯一的建议是 scheduler 和 worker 分开跑不要混在一个进程里否则调度器的数据库操作可能会被某个跑得特别久的任务拖住。4.2 实测数据与运行结果我在一台 2 核 4G 的虚拟机上做了模拟压力测试没有用真实业务数据而是造了五百个归档任务和五百个检查任务混跑。worker 进程数设为 4单个任务平均执行时间大约 5 秒整体吞吐量大约每秒 20 个任务高峰期数据库锁等待没有明显增长。跑完一千个任务失败率约 0.8%基本都是故意注入的异常任务重试后全部成功清零。和之前直接用 crontab 的方案对比最大提升不是速度而是确定性。以前一个任务到底跑没跑只能看日志时间戳现在直接查一张表就能列出来。另一个明显改善是任务之间的耦合变低了以前脚本里经常出现 A 脚本判断 B 脚本的结果文件是否存在现在我们把这种依赖关系改成“B 成功后才触发 A”配置里加一个 depends_on 字段就能表达确实少了很多玄学问题。不过我也要泼一点冷水这个方案在上千任务规模里表现很好但如果单任务本身就跑了几个小时worker 池再大也没用瓶颈会变成单任务自身逻辑。所以这类系统解决的是编排问题不是任务性能问题。任何调度工具都替代不了插件内部代码质量。5. 常见问题与排查技巧5.1 高频问题速查表问题现象可能原因排查思路解决办法任务一直 pending不进入 running调度器没启动或 next_run_at 未到查 scheduler 日志和任务表里的 next_run_at确认调度器在跑检查 cron 表达式任务卡在 running 状态worker 崩溃或心跳没更新看 heartbeat_at 是否很久没变配置心跳超时回收把状态重置为 pending任务重复执行缺少幂等键或状态抢占逻辑写错查 execution_id 是否唯一用先 UPDATE reserved 再执行的方式取任务执行过程中 SQLite 报 database is locked并发写太频繁没有开启 WAL查看 SQLite 版本和 journal_mode开启 WAL设置 busy_timeout重试次数超过配置但任务还是 dead插件内部有全局状态污染检查插件函数是否依赖全局变量确保插件是纯函数输入输出通过参数传递告警没收到Webhook 地址错误或网络不通先手动 POST 测试告警接口加失败重发机制告警本身也要可观测这张表里的问题很多都是初期频繁遇到的。尤其是 SQLite 锁一旦任务并发量上来总会冒出来。我的经验是不要一开始就用多进程一起写同一个 SQLite 文件要设计好“谁负责写状态”“谁负责读状态”。在 Apollo-11 里只有 scheduler 进程做状态写入worker 进程只上报心跳和结果这样能把锁竞争降到最低。5.2 亲手踩过的三个坑第一个坑是超时控制。我最初用 Python 的线程来执行每个任务超时就扔一个异常进去结果有些插件内部不响应线程中断任务明明超时了却还在跑。后来改成用 multiprocessing 进程隔离每个 worker 子进程执行一个任务超时后直接 terminate这才真正做到“超时即结束”。代价是进程启动有开销但对于分钟级任务完全可接受。第二个坑是日志和任务状态混在一起。早期我把任务日志统一打印到标准输出再去翻进程日志结果任务一多根本分不清哪段输出属于哪个任务。后来我把每次执行的标准输出单独存到一张 task_log 表里字段包括 execution_id、task_id、运行阶段、时间戳、内容摘要。这样出问题后直接用执行批次号过滤日志定位速度快非常多。日志输入输出本身也要异步不能阻塞任务主流程。第三个坑是系统时区。最开始调度器直接在 Python 里用datetime.now()结果到了跨天的时候有些任务莫名其妙提前或延后一小时。排查半天发现是服务器时区没有统一而 cron 表达式判断依赖本地时间。我后来把所有时间统一成 UTC 存储只在显示和配置层转成指定时区。这个改动虽然小但直接消除了所有因为时区导致的任务漏跑误跑。6. 一点反思和后续想做的事6.1 这类项目真正难在哪Apollo-11 从第一版能跑通到现在稳定运行中间迭代了很多轮。回头看最难的不是队列、不是调度循环而是状态一致性和失败恢复。很多任务系统表面看起来能用但一旦进程中途被杀掉、网络闪断、数据库突然发锁各种边缘情况会一起出现。如果没有前期把状态机设计清楚后面补丁会越打越多最后整个流程沦为“碰运气”。另一个难点是让插件作者包括未来的自己遵守约定。我踩过几次坑之后在文档里明确写了几条规则插件函数必须接受 params 字典必须返回一个可 JSON 化的结果必须自己保证业务幂等禁止读写全局状态。规则越简单越容易执行。如果你计划写类似工具我强烈建议一开始就把插件约束写到 README 里甚至在注册时自动做检查。6.2 后续扩展方向目前 Apollo-11 还缺几个部分。一个是 Web 界面现在只能查库体验还是不够直观后续我打算加一个只读面板展示队列长度、今日成功失败数、任务耗时分布。另一个是 worker 的跨机器注册现在 worker 只能跑在同一台机器上如果后续任务量变大我希望能把 worker 部署到多台机器通过租约机制领取任务而不需要整体改成分布式调度系统。如果你只是想把个人脚本统一管起来我的建议是不要过度模仿重型系统的设计。先把“任务”这个抽象做到完整把状态机、心跳、重试、幂等、告警这五件事想清楚你的工具就已经能解决绝大多数批处理问题。规模没那么大的时候简单系统带来的可控性比一堆炫酷的分布式组件更能救你于水火。最后分享一个我自己的习惯每一次调整调度逻辑之后我会刻意人为杀掉 worker 进程或者手动把某个任务状态改成 running然后观察系统能不能在几分钟内自我恢复。这种“破坏性演练”能逼出很多平时发现不了的边界问题。Apollo-11 能稳定跑这么久很大程度上要归功于这些刻意制造的混乱。如果你也在做类似的项目不妨也试试这个办法。