在反复写同一套“读配置、跑批、收结果”的脚本之后我大概是第三遍把同样的代码复制进新项目时终于决定把它抽出来做成一个自己的小库。这篇文章要聊的就是XSF 自制库——一套围绕“配置驱动的批处理与任务编排”搭起来的轻量工具集。它不解决什么宏大问题就是把那些散落在各种项目里的重复动作收口成一份可复用的代码一个 JSON 或 TOML 配置就能描述清楚要跑什么、按什么顺序跑、失败怎么办、结果汇总到哪。它适合两类人一类是长期靠零散脚本干活、已经被“改一处忘三处”折磨过的个人开发者另一类是团队里想统一小工具写法、但又不想为这点需求引入一整套重型编排框架的人。下面从它为什么长成这个样子讲起一路讲到下载、配置、跑通第一个示例以及我在实际使用中踩出来的那些坑全部按我自己的实践节奏来写。1. XSF 自制库的定位与整体设计1.1 先定边界这个库明确不做什么做自制库最容易犯的错是一开始雄心勃勃最后做成一个半成品的“平台”。我在动第一行代码之前先给自己画了三条红线这三条决定了 XSF 后面所有的结构。第一不做调度服务。它没有常驻进程没有任务队列没有 Web 界面。你在命令行敲一下它把你配置里列的任务按依赖顺序跑完打印汇总进程退出。想定时执行交给系统自带的定时任务工具。这个取舍让整个库的体积控制在一个目录里就能翻完的程度出问题的时候我可以直接读源码定位而不是去查一份几十页的文档。第二不做分布式。所有任务都在本机跑通过子进程或线程并发。这个决定的代价是它撑不起几百个节点的大规模场景但换来的是零网络依赖、零中间件部署。我大部分实际需求其实就是“本机把二十个步骤按顺序跑一遍其中几个可以并行”分布式在这一层是纯粹的负担。第三不做抽象层套娃。任务定义就是一个字典加几个约定键不需要继承基类不需要实现接口不需要注册插件。这一点是我踩过坑之后才定下来的早前写过一版带抽象基类的版本结果每加一个新任务类型就要写一个类、注册一次、再改一处工厂函数最后代码量比不用库还多。方案依赖体积上手成本排查难度适合场景重型编排框架大通常带一串传递依赖需要理解概念模型出错要深入框架内部多人协作、跨机器、有审计要求纯手写脚本无低低但重复一次性任务XSF单目录无第三方强依赖读一份配置格式即可出错位置基本落在自己写的任务里本机批处理、多步骤流水、个人与小组复用为什么最终选“配置 注册表”而不是“面向对象的任务类”理由很实际。我写脚本的时候脑子的模型是“先干这个再干那个这个失败了就重试两次”这是一条线性的叙述不是一个类图。配置文件的顺序天然就是这个叙述顺序改起来是改一行字不是改一个类的继承关系。人在赶时间的时候改一行字的成本远低于改一个类。1.2 四个核心模块怎么分工XSF 内部切成四块依赖方向严格单向配置层 → 注册层 → 执行层 → 汇总层。上层可以调用下层下层绝不回头引用上层。这条纪律看起来教条但它是我后来能放心改任何一个模块而不炸掉其他的根本原因。配置层负责把 JSON/TOML 读成一个内部对象做字段校验、默认值填充、路径规范化。它的输出是一个任务列表每个任务是一个标准化的字典。这一层最重要的设计是默认值要足够聪明超时不填就按 300 秒算重试次数不填就是 0工作目录不填就用配置文件所在目录。默认值定得好示例能短到十几行定得差用户第一件事就是去抄一份二十行的模板。注册层是一个字符串到可调用对象的映射表。XSF 自带几个内置动作执行命令、执行 Python 函数、调用 HTTP 接口、拷贝文件你也可以用一行装饰器把自己的函数挂进去。为什么用字符串键而不是直接传函数对象因为配置是纯数据必须能被序列化、能被外部生成、能被 diff。如果配置里塞的是函数引用这套东西就只能活在 Python 进程里跨语言调用和配置生成就全废了。执行层处理依赖拓扑排序、并发控制、超时、重试、失败传播。这是整个库里唯一有“算法味”的地方也是最容易出隐蔽 bug 的地方后面第 4 节会详细讲参数怎么算。汇总层收集每个任务的开始时间、结束时间、退出状态、标准输出尾部若干行、重试次数最后打一张表。这张表的用处比想象中大批处理跑完之后人往往只想知道“哪个失败了、失败在哪一行”而不是去翻几千行滚动日志。1.3 为什么不用“一个大函数”搞定有人会问这些功能用一个两百行的函数不也能凑出来吗能但会丢三样东西。一是可测性。拆成四层之后配置解析可以单独测拓扑排序可以单独测重试策略可以单独测。我之前那个大函数版本改一个重试逻辑要靠“跑一遍真实场景看结果对不对”一次验证五分钟。二是错误定位。分层之后错误会带着层级信息抛出来配置错就是配置错带字段名和行号任务错就是任务错带任务名和退出码。混在一起的版本里一个 KeyError 既可能是配置缺字段也可能是你自己的任务代码写错了排查全靠猜。三是复用粒度。汇总层那张表我在三个不同的项目里都单独 import 出来用过。 monolithic 的函数做不到这件事。2. 下载与版本选择拿到一份干净可用的包2.1 发布包里到底装了什么XSF 的发布包刻意做得很“扁”解压之后不需要安装步骤就能直接跑。目录结构大致是这样xsf-0.9.3/ ├── xsf/ │ ├── __init__.py │ ├── config.py # 配置加载与校验 │ ├── registry.py # 动作注册表 │ ├── runner.py # 依赖排序、并发、重试 │ ├── report.py # 结果汇总与打印 │ └── actions/ # 内置动作 │ ├── shell.py │ ├── python_call.py │ ├── http.py │ └── fs.py ├── examples/ │ ├── minimal.json │ └── pipeline.json ├── tests/ ├── CHANGELOG.md ├── LICENSE └── xsf.py # 单文件入口方便直接运行每个部分的作用和“能不能删”我列个表方便你拿到包之后做减法路径作用是否必需备注xsf/核心包必需所有逻辑都在这里xsf/actions/内置动作建议保留只删不用的动作文件也行但省不了多少空间examples/示例配置可删建议先留着第一条命令就是跑它tests/自测用例可删改源码时建议保留改完跑一遍能挡掉大部分低级错误xsf.py单文件入口可选不想装包、只想扔一个文件进项目的场景用它CHANGELOG.md变更记录建议保留升级前扫一眼能省掉很多“怎么突然报错了”的时间下载之后我建议第一件事是校验哈希。自制库的发布包通常不走大型包管理器的签名体系中间被替换或下载不完整是有可能的。发布页一般会给出 SHA256本地算一遍对一下# 计算下载文件的哈希 sha256sum xsf-0.9.3.zip # 输出示例对照发布页给出的值 # 7f3c9a1e5b2d4c8f0a6e1b3d9c7f2a4e8b6d0c3f5a1e9b7d2c4f6a8e0b1d3c5f xsf-0.9.3.zip对不上就别解压重新下一遍。这一步看起来很形式主义但我确实遇到过一次网络中断导致压缩包少了尾部字节解压时报的错是“CRC 校验失败”如果当时没做哈希校验我大概率会以为是解压工具的问题去折腾一圈工具。2.2 版本号怎么挑0.x 阶段的兼容性现实XSF 目前还在 0.x 阶段这个阶段按语义化版本的约定次版本号变化可能带来不兼容改动。我把过去几个版本的关键变化整理成一张表你在选版本的时候可以直接对照版本配置文件格式关键变化兼容性提示0.6.xJSON only基础任务链无depends_on顺序即执行顺序0.7.xJSON only引入depends_on与拓扑排序老配置可直接用但不享受并行0.8.xJSON / TOML引入retry_backoff字段retries语义从“总尝试次数”改为“额外重试次数”0.9.xJSON / TOML引入fail_fast与汇总表默认值改为fail_fasttrue与旧版行为相反看到 0.8.x 那一行的坑了吗retries的语义改过一次。老配置写retries: 2表示“最多跑两次”新版本里同样的写法表示“失败后再试两次最多跑三次”。这类语义漂移是 0.x 阶段最难防的问题因为它不报错只是行为悄悄变了。我的建议是配置里永远显式写全关键字段不要依赖默认值。多写几个字换来的升级时不用提心吊胆。另外在项目里固定版本号不要用“取最新”的方式引用否则某天早上起来流水线行为变了你会花半天时间去怀疑自己的代码。2.3 三种放置方式按你的项目形态选拿到包之后怎么放取决于你用它的场景。方式一直接解压到项目子目录。适合“我就是这个项目里用一下”的情况。把xsf/整个目录拷到项目根下代码里import xsf就能用。好处是零安装、跟着项目走、clone 下来就能跑。坏处是多个项目各有一份改了一处其他地方不会同步。方式二装成本地包。适合“我有五六个项目都要用想统一升级”的情况。在包目录里执行本地安装# 在解压后的目录内执行安装到当前 Python 环境 python -m pip install . --no-deps # 验证安装 python -c import xsf; print(xsf.__version__)--no-deps是我习惯加的因为 XSF 核心不依赖第三方库加了反而可能触发一些不必要的解析。如果你用的版本引入了可选依赖比如 TOML 解析在某些老 Python 上需要额外库去掉这个参数让它自己装。方式三只放单文件入口。适合“临时跑一次不想污染环境”。把xsf.py拷到任意目录用python xsf.py run config.json直接跑。这个单文件是把核心逻辑打平后的版本功能略少不支持自定义动作注册但足够应付简单场景。注意三种方式不要混用。同一个 Python 环境里既装了包、项目目录下又有一份同名目录import的优先级会让你拿到哪一份完全取决于当前工作目录这类问题排查起来极其费时。3. 配置详解把字段一个一个说清楚3.1 配置文件长什么样每个字段管什么先看一份完整的配置然后逐个字段拆。这是examples/pipeline.json的稍加改动版{ name: daily-batch, defaults: { timeout: 300, cwd: ., env: { XSF_LOG_LEVEL: info } }, tasks: [ { name: fetch, action: shell, params: { command: python tools/fetch_data.py --date ${DATE} }, timeout: 600, retries: 2, retry_backoff: 5 }, { name: clean, action: python, params: { target: tools.clean:run, args: [--strict] }, depends_on: [fetch] }, { name: report, action: shell, params: { command: python tools/build_report.py }, depends_on: [clean], fail_fast: false } ] }字段含义我做成表格配置写错的时候可以直接对照查字段层级类型默认值作用与注意点name顶层string无必填批次名字只用于日志和汇总表标题defaults顶层object{}全局默认值任务里同名字段会覆盖它defaults.timeout默认number300单个任务超时秒数超时按失败处理defaults.cwd默认string配置文件所在目录任务工作目录相对路径以配置文件位置为基准defaults.env默认object继承当前环境追加到子进程环境变量不是替换tasks[]顶层array无必填任务列表顺序不影响执行依赖决定顺序tasks[].name任务string无必填全局唯一重复会直接报配置错误tasks[].action任务string无必填内置动作名或自己注册的动作名tasks[].params任务object{}传给动作的参数内容由动作自己解释tasks[].timeout任务number继承默认秒。长任务一定要显式调大别用默认值赌tasks[].retries任务integer0额外重试次数不是总次数tasks[].retry_backoff任务number0首次退避秒数实际等待按指数增长tasks[].depends_on任务array[]前置任务名列表决定执行顺序与并行度tasks[].fail_fast任务booleantruetrue 时失败立即终止全批false 则继续跑无关任务tasks[].skip_if任务string无条件表达式为真时跳过该任务并标记为 skipped关于fail_fast我想多讲两句。这个字段我起初觉得没什么用直到有一次批处理里有“生成数据”和“清理临时文件”两个任务前者失败之后后者如果也被终止临时目录会越积越大第二天磁盘就报警了。把清理任务的fail_fast设成false并且不依赖那个失败任务它就能在别的任务炸掉之后依然跑完。还有env这个字段容易被误解。它是追加而不是替换也就是说子进程能看到父进程已有的环境变量再加上你写在这里的。这个设计是刻意的因为替换式环境会让PATH之类的变量消失导致命令找不到。但反过来说如果你想清掉某个变量得写成空字符串而不是删掉这一行。3.2 变量替换与路径规则配置里支持${VAR}形式的占位符来源有两个一是defaults.env里的键二是启动时传入的--var参数。这个功能我一开始没加后来发现每次换日期都要手改配置实在受不了。# 启动时注入变量 python -m xsf run config.json --var DATE2024-06-01 --var REGIONcn-north替换规则有三条记住就行${VAR}出现在params的字符串值里会被替换出现在键名里不会。未定义的变量会直接报错而不是替换成空串。这是故意的替换成空串会让命令变成--date看起来像跑成功了实际上跑的是错的数据。想要字面量的${写成$${。路径规则是另一个高频出错点。cwd的基准是配置文件所在的目录不是执行命令时的当前目录。这个设计让配置具有可移植性config.json和它的相对路径依赖一起移动不会失效。但如果你习惯在项目的某个子目录里执行命令第一次用会觉得“怎么找不着文件”。我的做法是不管在哪执行一律写cwd显式指定把不确定性降到零。4. 简易示例与参数调优4.1 最小可运行示例十几行跑通先跑最小的一份确认环境没问题。下面这个文件存成hello.json{ name: hello, tasks: [ { name: say, action: shell, params: { command: python -c \print(xsf ok)\ } } ] }执行python -m xsf run hello.json预期输出大致是这样时间戳和耗时因人而异[xsf] batch: hello [xsf] plan : say [xsf] run : say ... ok (0.12s) [xsf] ---- summary ---- [xsf] total 1 | ok 1 | fail 0 | skipped 0 | elapsed 0.12s跑通这一步之后把它改成真实的三步流水。这里用 Python 代码而不是命令行是因为我想演示自定义动作注册这个更常用的形态# run_pipeline.py import xsf # 注册一个自定义动作名字叫 process xsf.action(process) def process(payload, ctx): # payload 就是配置里的 params # ctx 里有 vars、logger、task_name threshold int(payload.get(threshold, 10)) rows ctx.vars.get(ROWS, 100) ctx.logger.info(processing rows%s threshold%s, rows, threshold) if int(rows) threshold: raise ValueError(frows {rows} below threshold {threshold}) return {processed: int(rows)} if __name__ __main__: xsf.run(pipeline.json, vars{ROWS: 500})对应的pipeline.json{ name: demo, defaults: { timeout: 120 }, tasks: [ { name: step_a, action: process, params: { threshold: 100 }, retries: 1 }, { name: step_b, action: process, params: { threshold: 50 }, depends_on: [step_a] } ] }step_b会在step_a成功之后才开始。如果step_a抛异常它会按retries: 1再试一次还是失败就整批中断因为默认fail_fast是 true。4.2 并发数、超时、退避这三个参数怎么算这三个参数如果拍脑袋填要么跑得慢要么把机器压垮要么在网络抖动的时候误判失败。我把自己算它们的方法写下来你可以直接套。并发数。XSF 用的是线程池 子进程的方式实际并发受两件事限制CPU 核数和 IO 等待比例。经验公式是建议并发数 N min(CPU 核数 × (1 平均IO等待时间 / 平均计算时间), 任务总数)举个具体例子。我这台机器 8 核跑的是“下载 解析”类任务单个任务平均耗时 4 秒其中真正吃 CPU 的大概 0.5 秒剩下 3.5 秒在等网络。那么1 3.5/0.5 88 × 8 64看起来能开到 64。但实际受限于对端服务我实测下来稳定在16 左右再往上错误率明显上升重试反而拖慢了整体。所以公式算出来的是理论上限实际值要靠一轮压测收敛。并发数20 个任务总耗时失败率备注421.3s0%稳但慢811.6s0%日常首选167.4s0.5%性价比拐点327.1s6.2%重试吃掉收益648.9s21%明显过载这张表是我在自己环境里用 20 个性质相近的任务测出来的你的绝对值肯定不一样但拐点形状大概率类似先线性下降然后因为对端限流开始反弹。找拐点的方法是二分从 4 开始翻倍跑到耗时不再明显下降为止取前一个值。超时时间。别用“感觉差不多”的数字。正确做法是先跑一轮统计每个任务耗时的 P99然后设成P99 × 1.5。为什么要乘 1.5 而不是乘 1.1因为超时的代价是不对称的设太松一个卡死的任务会拖住整条链设太紧正常的慢请求被判死触发无谓重试反而更慢。1.5 倍是留出正常波动空间又能在异常时较快兜底的折中值。P99 118s → timeout 118 × 1.5 ≈ 177s取整 180s重试退避。XSF 的退避是指数增长第 n 次重试等待retry_backoff × 2^(n-1)秒。retry_backoff: 5、retries: 3的等待序列是 5s、10s、20s累计 35 秒。这个设计针对的是“临时性故障”——对端短暂不可用、文件被别的进程占着。如果是确定性错误参数写错、文件不存在重试纯粹是浪费时间这时候retries应该设 0让它快速失败。我的经验配置是网络类任务retries: 2, retry_backoff: 5本地文件类任务retries: 1, retry_backoff: 1纯计算类任务retries: 0失败就修代码。注意retries的次数是按“任务”算的不是按“命令里的每一步”算。一个任务里如果串了三条命令重试是把三条命令整个再跑一遍前面的副作用会重复发生。涉及写文件的任务务必把写入做成幂等的。5. 常见问题与排查技巧实录5.1 报错速查表下面这些是我自己和小范围试用时反复遇到的按现象分类现象大概率原因解决方式ConfigError: unknown field retry字段名拼错正确是retries拼写检查配置解析是严格模式容错字段不会静默忽略ConfigError: duplicate task name两个任务同名重命名同名不会覆盖直接报错CycleDetected: a - b - adepends_on成环画出依赖关系删掉多余那条边任务一直不结束最后超时子进程在等标准输入命令里加非交互参数如--yes或重定向 /dev/null拿不到任务输出输出量过大被截断汇总表只保留尾部若干行完整输出看日志文件FileNotFoundError找不到脚本cwd基准不是当前目录显式写cwd或把路径改成相对配置文件Windows 下命令报“不是内部或外部命令”用了 shell 内建命令但没走 shell显式指定shell: true或改成调用可执行文件重试次数感觉不对把retries当成总次数它是额外重试次数总执行次数 retries 1环境变量没生效写在了任务层但被默认层覆盖检查defaults.env与任务层同名键任务层优先中文输出乱码子进程编码与终端不一致显式设PYTHONIOENCODINGutf-8或统一用 UTF-85.2 几条踩坑心得第一条子进程不清理会积累僵尸。XSF 执行 shell 动作时会起子进程。如果超时被强杀只杀父进程而不杀进程组被它拉起来的孙子进程会挂在后台。我早期版本就吃过这个亏跑了一晚上之后机器上挂着两百多个孤儿进程。修复方式是启动时把子进程放进独立进程组超时的时候对整个组发终止信号。如果你自己写类似逻辑这条务必记住。第二条日志要带任务名而且要在开始时打一行。汇总表再好用也只是事后的。任务卡住的时候你最想知道的是“现在跑到哪一步了”。XSF 在任务开始时立刻打一行run: name失败时补一行带退出码的。这个设计看起来简陋但实际排查时比什么都管用——一眼就知道卡在哪个任务。第三条不要用配置文件写敏感信息。配置是要进版本库、要被 diff、要被分享的。密码、令牌这类东西通过--var或者环境变量注入配置里只写${TOKEN}这样的占位符。这不是为了安全级别多高而是为了让“配置能不能给人看”这个问题永远有肯定答案。第四条自定义动作里的异常要老实抛出去。XSF 的动作函数返回什么不影响流程抛不抛异常才影响。我见过有人在动作里try/except把所有异常吞掉然后返回一个{ok: False}结果外层认为任务成功了fail_fast永远不触发。要么抛异常要么用一个明确的检查步骤去判断别让失败伪装成成功。第五条先用串行跑通再开并发。并发带来的问题顺序依赖、共享文件竞争、对端限流会掩盖掉任务本身的 bug。我的习惯是--jobs 1跑三遍确认稳定再逐步往上加。这个顺序反过来做你会花大量时间在“到底是我的任务写错了还是并发的问题”之间反复摇摆。第六条给配置加一个--dry-run习惯。正式跑之前先看一眼计划确认任务顺序和依赖符合预期。牵涉到删除、覆盖这类不可逆操作的时候这一步能救你。XSF 的 dry-run 只打印拓扑排序结果不执行任何动作成本几乎为零。第七条任务粒度别切太细。刚开始用的时候容易把每个命令都拆成一个任务结果配置比脚本还长。我的经验是有独立失败语义和独立重试需求的才拆成任务纯粹的顺序命令合并到一个 shell 动作里就好。判断标准很简单——如果这个步骤失败了我会不会想单独重试它会就拆不会就合。第八条版本升级前先读 CHANGELOG特别是 0.x 阶段。前面提过retries语义变更那次我是靠一个跑了三个月的老配置行为异常才发现的中间浪费了一整个下午。现在我的习惯是升级前扫一眼变更记录把涉及的字段在自己配置里 grep 一遍改完再升。多花五分钟省下半天。最后再分享一个小技巧。XSF 的汇总表默认只打终端但我在长时间批处理里会给它加一个--report-json out.json参数把结果同时写成 JSON。原因很实在批处理跑完之后我经常是第二天早上才看终端早就滚没了而out.json可以被另一个小脚本读去做飞书或邮件通知。这个「输出结构化结果」的习惯我从这个库一直沿用到所有自建工具上回报率比想象中高得多。