1. 从“rea”这个标题说起一个被低估的缩写背后藏着什么第一次看到“rea”这个标题的时候我承认自己愣了几秒。没有正文没有关键词没有摘要就孤零零三个字母。这种极简到近乎“空白”的输入反而让我觉得有意思——因为在实际工作中我们经常遇到类似的情况一个模糊的需求、一个没头没尾的代号、一个别人嘴里蹦出来的缩写然后你得靠自己的经验去把它落地。“rea”在不同语境下可以指向完全不同的东西。做前端的人第一反应可能是React的某种简写习惯做硬件的人会想到Read-Eval-Assert这类测试循环做图像处理的人会联想到Real-time Embedded Architecture做音频的人会想到Reverberation Enhancement Amplifier。但结合“最新网络热词”这个提示我更倾向于把它理解成一个轻量级、可复用的自动化执行框架的代号——因为“rea”读起来干脆利落很适合做工具名而且三个字母的组合在命名空间里足够短方便在命令行里敲。所以这篇博文我打算围绕“如何从零搭建一个代号为rea的轻量级任务执行器”来展开。它不是一个具体的开源项目而是一个设计思路实操落地的完整记录。适合那些手头有一堆重复性操作、想用最小成本做自动化、又不想引入重型框架的开发者。读完你至少能拿到一套可运行的骨架代码以及我在搭建过程中踩过的五六个坑。提示本文所有代码和配置都基于通用技术栈不依赖任何特定云服务或商业产品你可以直接复制到本地环境跑通。2. 为什么我选择用“rea”做一套自己的任务执行器2.1 现成方案那么多为什么还要自己写市面上任务队列和自动化工具确实不少从重量级的分布式调度平台到轻量级的脚本管理器选择很多。但我在实际项目里发现一个尴尬的中间地带用大框架太重用裸脚本太乱。比如你只是想让几个数据清洗步骤按顺序跑中间某个步骤失败了能重试跑完发个通知——这种需求用Airflow之类的工具光环境搭建和依赖管理就够喝一壶但如果你直接写一个bash脚本错误处理、日志记录、状态追踪又得自己从头造。“rea”的定位就是填这个空档。它的核心目标只有三个顺序执行、失败重试、结果可查。不搞分布式不搞可视化编排不搞复杂的依赖图。一个配置文件描述任务链一个命令行入口启动执行一个本地文件记录每次运行的状态。就这么简单。我试过用现成的轻量级库但要么是API设计太啰嗦要么是扩展性差——想加一个自定义的失败通知渠道得翻半天源码。自己写反而更可控代码量也不大核心逻辑大概三百行就能搞定。2.2 “rea”这个名字的由来与设计约束名字其实没什么玄机就是“run-execute-archive”三个单词的首字母。Run是启动Execute是执行具体步骤Archive是归档结果。三个字母刚好构成一个完整的生命周期读起来也顺口。设计约束我给自己定了四条零外部依赖除了语言标准库不引入任何第三方包。这样在任何干净环境里都能跑起来不用折腾虚拟环境或包管理器。配置即代码任务链用YAML或JSON描述但允许在步骤里嵌入脚本片段。这样既保持了配置的可读性又不用为了一个简单逻辑去写插件。状态持久化到本地文件每次运行生成一个带时间戳的目录里面包含日志、输出文件和最终状态标记。不依赖数据库方便直接查看和清理。失败可重入如果一个任务链在中途失败重新运行时能从失败的那一步继续而不是从头再来。这个特性在调试阶段特别省时间。这四条约束决定了整个架构的走向。比如“零外部依赖”意味着我不能用现成的重试库得自己写一个带指数退避的循环“状态持久化到本地文件”意味着并发控制要靠文件锁来实现而不是数据库事务。2.3 一个典型的使用场景长什么样假设你每天需要做这样一件事从某个数据源拉取原始文件做格式转换然后跑三个校验脚本最后把结果打包上传到指定目录。手动做的话你得记住每一步的命令和参数中间某一步报错了还得回头翻日志。用rea的话你只需要写一个这样的配置文件name: daily-data-pipeline steps: - name: fetch command: python fetch.py --date {{today}} retry: 3 timeout: 300 - name: transform command: python transform.py --input ./raw --output ./clean depends_on: fetch - name: validate-schema command: python validate.py --mode schema depends_on: transform - name: validate-range command: python validate.py --mode range depends_on: transform - name: package command: tar -czf result.tar.gz ./clean depends_on: [validate-schema, validate-range]然后执行rea run daily-data-pipeline.yaml剩下的交给它。每一步的输出会写到独立的日志文件失败会自动重试全部完成后会在归档目录里生成一个SUCCESS标记文件。如果中途失败了你改完脚本再跑一次它会从失败的那一步继续。这种场景在数据工程、运维巡检、甚至日常的文件整理里都很常见。rea不解决复杂问题它只是把“一串命令按顺序跑”这件事做得更可靠一点。3. 核心模块拆解一个执行器到底需要哪几块骨头3.1 配置解析层YAML还是JSON以及变量替换的坑配置格式我最终选了YAML原因很简单写多行命令的时候不用处理转义可读性好。但YAML有个著名的坑——缩进必须用空格不能用Tab。我在早期版本里没做校验结果用户从网页复制配置过来里面混了Tab解析直接报错错误信息还特别模糊。后来我在解析前加了一步预处理把所有Tab替换成两个空格然后再交给YAML解析器。变量替换是另一个容易出问题的地方。我支持两种语法{{today}}这种双花括号的简单替换以及${ENV_VAR}这种环境变量引用。简单替换的值来自内置的上下文对象比如当前日期、运行ID、上一步的输出路径等。环境变量引用则直接读系统环境。这里有个细节替换发生在命令执行之前但不在配置解析阶段。也就是说配置里的{{today}}在解析时保持原样等到真正要执行某一步的时候才替换成实际值。这样做的好处是如果某一步失败了你重新运行时{{today}}会重新计算而不是沿用上次的旧值。import re import os from datetime import datetime def resolve_variables(text, context): # 先处理 {{var}} 形式 def replace_curly(match): key match.group(1).strip() return str(context.get(key, match.group(0))) text re.sub(r\{\{(.?)\}\}, replace_curly, text) # 再处理 ${VAR} 形式 def replace_env(match): key match.group(1).strip() return os.environ.get(key, match.group(0)) text re.sub(r\$\{(.?)\}, replace_env, text) return text注意环境变量替换不要用os.path.expandvars因为它对未定义的变量会保留原样而我希望未定义的变量直接报错避免静默失败。3.2 执行引擎顺序、并行与依赖判断执行引擎的核心是一个拓扑排序加状态机。每个步骤有三种状态pending、running、done或failed。引擎从所有pending的步骤里找出依赖已经满足的然后按配置顺序依次执行。依赖判断的逻辑很简单一个步骤的所有depends_on列表里的步骤都处于done状态它才可以被调度。如果某个依赖是failed那么这个步骤直接标记为skipped不再执行。我一开始想支持并行执行但后来放弃了。原因是并行会引入资源竞争和日志交错的问题而rea的目标场景里步骤之间通常有数据依赖真正能并行的不多。强行并行反而增加了复杂度和调试难度。所以最终版本是严格顺序执行但依赖判断仍然保留用来处理“多个步骤依赖同一个前置步骤”的情况。def get_ready_steps(steps, status): ready [] for step in steps: if status[step[name]] ! pending: continue deps step.get(depends_on, []) if isinstance(deps, str): deps [deps] if all(status.get(d, pending) done for d in deps): ready.append(step) return ready执行一个步骤时我会用subprocess.run启动子进程并设置超时。超时时间从配置里读默认是600秒。如果超时子进程会被终止该步骤标记为失败。3.3 状态存储为什么我用文件而不是数据库状态存储我选择了最朴素的方式每次运行创建一个目录路径是./rea-runs/run_id/其中run_id是时间戳加随机后缀。目录里包含config.yaml本次运行的配置快照status.json每个步骤的当前状态和开始/结束时间logs/每个步骤的标准输出和标准错误outputs/步骤产生的文件如果配置了输出路径SUCCESS或FAILED最终标记文件用文件的好处是透明。你不需要任何工具就能查看运行历史直接ls和cat就行。清理也简单删目录即可。坏处是并发写入需要加锁。我的做法是在修改status.json之前先获取一个文件锁用fcntl.flock写完释放。这样即使多个rea进程同时操作同一个运行目录虽然不推荐也不会把状态写乱。import fcntl import json def update_status(run_dir, step_name, new_status): status_file os.path.join(run_dir, status.json) with open(status_file, r) as f: fcntl.flock(f, fcntl.LOCK_EX) data json.load(f) data[steps][step_name][status] new_status data[steps][step_name][updated_at] datetime.now().isoformat() f.seek(0) f.truncate() json.dump(data, f, indent2) fcntl.flock(f, fcntl.LOCK_UN)提示fcntl是Unix系统特有的如果你在Windows上跑需要用msvcrt.locking替代。为了保持跨平台我后来把锁的实现抽象成了一个函数根据sys.platform选择不同的底层调用。3.4 日志与归档让每一次运行都可追溯日志我分了两级步骤级日志和运行级日志。步骤级日志就是每个步骤的stdout和stderr分别写到logs/step_name.out和logs/step_name.err。运行级日志是一个总的run.log记录引擎自身的动作比如“开始执行步骤X”、“步骤X失败重试第2次”、“所有步骤完成”。归档策略是运行结束后如果所有步骤都成功在运行目录里创建一个SUCCESS空文件如果有任何步骤失败创建FAILED文件并把失败步骤的名称和错误码写进去。这样你写一个简单的shell脚本就能监控运行结果if [ -f ./rea-runs/latest/SUCCESS ]; then echo Pipeline completed successfully else echo Pipeline failed, check logs fi我还加了一个rea clean命令用来删除超过N天的运行目录。默认保留最近7天可以通过--keep-days参数调整。这个命令在CI环境里特别有用避免磁盘被历史运行记录撑满。4. 动手实现从空目录到能跑通第一条任务链4.1 环境准备与项目骨架我用的语言是Python 3.8因为标准库足够丰富而且跨平台。项目结构如下rea/ ├── rea/ │ ├── __init__.py │ ├── cli.py # 命令行入口 │ ├── config.py # 配置解析 │ ├── engine.py # 执行引擎 │ ├── state.py # 状态管理 │ └── utils.py # 工具函数 ├── tests/ │ └── test_engine.py ├── setup.py └── README.md安装方式有两种直接pip install -e .做开发安装或者把rea目录加到PYTHONPATH里。我推荐前者因为命令行入口会自动注册。setup.py里关键的一行是entry_points{ console_scripts: [ rearea.cli:main, ], },这样安装后就能直接在终端敲rea命令了。4.2 配置文件格式定义与校验配置文件我用YAML顶层字段有字段类型必填说明namestring是任务链名称用于日志和归档目录命名stepslist是步骤列表至少一个varsdict否自定义变量可在命令里用{{var}}引用retry_defaultint否默认重试次数默认0timeout_defaultint否默认超时秒数默认600每个步骤的字段字段类型必填说明namestring是步骤名称同一任务链内唯一commandstring是要执行的命令支持变量替换depends_onstring或list否依赖的步骤名retryint否该步骤的重试次数覆盖默认值timeoutint否该步骤的超时秒数覆盖默认值envdict否额外的环境变量校验逻辑我写了一个validate_config函数检查必填字段、名称唯一性、依赖是否存在、是否有循环依赖。循环依赖检测用深度优先搜索如果发现回边就报错。def detect_cycle(steps): graph {s[name]: s.get(depends_on, []) for s in steps} for s in steps: deps s.get(depends_on, []) if isinstance(deps, str): graph[s[name]] [deps] visited set() stack set() def dfs(node): if node in stack: return True if node in visited: return False visited.add(node) stack.add(node) for dep in graph.get(node, []): if dfs(dep): return True stack.remove(node) return False for node in graph: if dfs(node): return True return False4.3 执行循环的代码实现与关键细节执行循环的伪代码大概是这样加载配置 - 初始化运行目录 - 写入初始状态 while 还有pending步骤: 找出所有依赖满足的pending步骤 if 没有可执行的步骤: 标记所有剩余pending为skipped break 取第一个可执行步骤 更新状态为running 执行命令带重试和超时 if 成功: 更新状态为done else: 更新状态为failed 标记所有依赖它的步骤为skipped 跳出循环 写入最终标记文件重试逻辑我用了指数退避第一次失败后等1秒第二次等2秒第三次等4秒以此类推但最大不超过30秒。这样在遇到临时性故障比如网络抖动时重试的成功率会高很多。import time def execute_with_retry(command, retry, timeout, env): attempt 0 while attempt retry: try: result subprocess.run( command, shellTrue, timeouttimeout, capture_outputTrue, textTrue, envenv ) if result.returncode 0: return True, result.stdout, result.stderr else: attempt 1 if attempt retry: wait min(2 ** (attempt - 1), 30) time.sleep(wait) except subprocess.TimeoutExpired: attempt 1 if attempt retry: time.sleep(min(2 ** (attempt - 1), 30)) return False, result.stdout if result in dir() else , result.stderr if result in dir() else 注意subprocess.run的capture_output参数在Python 3.7才支持。如果你用更早的版本需要手动设置stdoutsubprocess.PIPE和stderrsubprocess.PIPE。4.4 跑通第一个示例一个三步数据处理链我写了一个最简单的示例来验证整个流程。任务链包含三个步骤生成数据、处理数据、校验结果。name: demo-pipeline steps: - name: generate command: python -c import random; open(data.txt,w).write(\n.join(str(random.randint(1,100)) for _ in range(10))) - name: process command: python -c linesopen(data.txt).readlines(); open(result.txt,w).write(str(sum(int(l) for l in lines))) depends_on: generate - name: verify command: python -c valint(open(result.txt).read()); assert 10 val 1000, out of range; print(OK) depends_on: process执行rea run demo.yaml输出大概是[2024-01-15 10:30:01] Starting run: 20240115-103001-a1b2 [2024-01-15 10:30:01] Step generate: running [2024-01-15 10:30:02] Step generate: done [2024-01-15 10:30:02] Step process: running [2024-01-15 10:30:03] Step process: done [2024-01-15 10:30:03] Step verify: running [2024-01-15 10:30:04] Step verify: done [2024-01-15 10:30:04] All steps completed. Run directory: ./rea-runs/20240115-103001-a1b2然后你可以在./rea-runs/20240115-103001-a1b2/里看到SUCCESS文件、status.json和各个步骤的日志。整个过程不到五秒但已经覆盖了配置解析、依赖调度、命令执行、状态记录和归档的全部环节。5. 踩坑记录那些文档里不会写的真实问题5.1 命令里的引号嵌套导致解析失败这个问题我遇到不止一次。用户在YAML里写命令时经常需要嵌套引号。比如command: python -c print(hello)YAML解析后这个字符串本身是python -c print(hello)没问题。但如果用户写成command: echo Its a testYAML会把Its里的单引号当成普通字符但如果你在命令里用了双引号包裹而命令本身又需要双引号就会出问题。更麻烦的是当我把命令传给subprocess.run(shellTrue)时shell会再解析一次引号导致实际执行的命令和预期不符。我的解决方案是在配置里尽量用单引号包裹整个命令内部需要引号的地方用双引号。如果实在需要嵌套建议把命令写成一个独立的脚本文件然后在配置里调用脚本。这样既避免了引号问题也让命令更易读。提示如果你非要在命令里用复杂引号可以用YAML的块标量语法command: python -c import json data {key: value} print(json.dumps(data)) 但要注意块标量会把换行符保留而shell对多行命令的处理方式取决于具体shell。我实测下来bash可以正确处理但某些精简shell比如dash可能会报错。5.2 子进程超时后残留进程的清理subprocess.run的timeout参数在超时后会杀死子进程但只杀死直接子进程不杀死它的后代进程。如果你的命令是bash -c python long_running.py超时后bash被杀了但python进程可能还在跑。这个问题在长时间运行的任务里特别危险会积累僵尸进程。我的修复方案是在Unix系统上启动子进程时设置start_new_sessionTrue这样会创建一个新的进程组。超时后用os.killpg杀死整个进程组而不是只杀父进程。import os import signal import subprocess def run_command(command, timeout): proc subprocess.Popen( command, shellTrue, stdoutsubprocess.PIPE, stderrsubprocess.PIPE, textTrue, start_new_sessionTrue ) try: stdout, stderr proc.communicate(timeouttimeout) return proc.returncode, stdout, stderr except subprocess.TimeoutExpired: os.killpg(os.getpgid(proc.pid), signal.SIGTERM) time.sleep(1) if proc.poll() is None: os.killpg(os.getpgid(proc.pid), signal.SIGKILL) return -1, , Timeout注意start_new_sessionTrue在Windows上不可用。Windows下需要用creationflagssubprocess.CREATE_NEW_PROCESS_GROUP然后用proc.send_signal(signal.CTRL_BREAK_EVENT)来终止。跨平台处理起来比较啰嗦但为了稳定性值得。5.3 状态文件并发写入的锁竞争前面提到我用fcntl.flock做文件锁。但在实际使用中我发现一个问题如果两个rea进程同时操作同一个运行目录比如用户不小心启动了两次锁竞争会导致其中一个进程长时间阻塞。更糟的是如果持有锁的进程崩溃了锁可能不会自动释放取决于操作系统导致另一个进程永远等下去。我的改进措施有两个一是在获取锁时设置超时用fcntl.flock(f, fcntl.LOCK_EX | fcntl.LOCK_NB)尝试非阻塞获取如果失败就等待一小段时间再试最多重试10次。二是在运行目录里放一个lock文件记录持有锁的进程ID和启动时间。如果发现锁文件存在但对应进程已经不存在了就强制清理锁文件。def acquire_lock(run_dir, timeout30): lock_file os.path.join(run_dir, lock) start time.time() while time.time() - start timeout: try: fd os.open(lock_file, os.O_CREAT | os.O_RDWR) fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB) os.write(fd, str(os.getpid()).encode()) return fd except BlockingIOError: time.sleep(0.5) raise TimeoutError(Could not acquire lock)5.4 变量替换时未定义变量的静默失败早期版本里如果配置里写了{{undefined_var}}而上下文里没有这个变量我的替换函数会保留原样导致命令里出现字面的{{undefined_var}}。shell执行时通常会报错但错误信息很难懂。更隐蔽的情况是如果这个变量出现在路径里可能会创建一个名字奇怪的目录而不是报错。后来我改成遇到未定义的变量直接抛出异常并在异常信息里指出是哪个变量、在哪个步骤里。这样用户能立刻发现问题而不是等到命令执行失败才去排查。def resolve_variables(text, context): def replace_curly(match): key match.group(1).strip() if key not in context: raise ValueError(fUndefined variable: {key}) return str(context[key]) return re.sub(r\{\{(.?)\}\}, replace_curly, text)这个改动虽然让配置更严格但省去了大量调试时间。我的经验是在自动化工具里静默失败比直接报错更可怕。宁可让用户一开始就写对也不要让他们在运行到一半时才发现问题。5.5 日志文件过大导致磁盘写满有一次我在一个循环任务里跑了上千次rea每次运行都会生成日志文件。虽然单个日志不大但累积起来很快就把磁盘写满了。问题出在我没有对日志做轮转或清理。修复方案是在rea clean命令里增加日志大小检查如果单个运行目录超过一定大小比如100MB就只保留status.json和SUCCESS/FAILED标记删除详细的步骤日志。另外在配置里增加一个log_max_size选项超过这个大小的日志会被截断只保留最后N行。def truncate_log(log_path, max_bytes10*1024*1024): if os.path.getsize(log_path) max_bytes: with open(log_path, rb) as f: f.seek(-max_bytes, 2) data f.read() with open(log_path, wb) as f: f.write(b... [truncated] ...\n) f.write(data)这个功能在长时间运行的任务里特别有用避免因为日志膨胀导致磁盘故障。6. 进阶玩法让rea适配更复杂的日常场景6.1 用钩子函数扩展通知渠道rea本身不内置通知功能但我留了一个钩子接口在配置里可以指定on_success和on_failure两个命令分别在任务链成功或失败后执行。这样你可以接任何通知方式比如发邮件、写消息队列、调用Webhook。name: pipeline-with-hooks on_success: curl -X POST https://example.com/notify -d {status:ok} on_failure: python send_alert.py --message Pipeline failed steps: - name: do-something command: echo working钩子命令的执行结果不影响主流程的状态。即使钩子本身失败了主流程的SUCCESS或FAILED标记也不会改变。这样设计是为了避免通知失败导致整个任务被误判。6.2 把rea嵌入CI流水线的最小实践在CI环境里我通常把rea作为构建步骤的一部分。比如在GitLab CI里stages: - test - deploy run-pipeline: stage: test script: - pip install -e . - rea run ci-pipeline.yaml artifacts: paths: - rea-runs/latest/ when: always关键点是when: always这样即使rea执行失败运行目录也会被保存为制品方便事后排查。另外我会在CI配置里设置REA_RUN_DIR环境变量把运行目录指向一个固定的路径而不是默认的./rea-runs这样制品收集更可控。6.3 多环境配置的变量覆盖策略同一个任务链在不同环境开发、测试、生产里可能需要不同的参数。我的做法是在配置里用变量占位然后通过环境变量或命令行参数覆盖。name: multi-env vars: db_host: localhost db_port: 5432 steps: - name: connect command: python connect.py --host {{db_host}} --port {{db_port}}运行时可以用rea run config.yaml --var db_hostprod.example.com --var db_port5433来覆盖。命令行参数的优先级高于配置文件里的vars这样既保持了配置的通用性又允许灵活调整。提示如果某个变量在所有环境里都必须不同建议不要写在配置文件里而是完全通过环境变量传入。这样配置文件可以纳入版本控制而敏感信息不会泄露。6.4 性能观察rea在长任务链下的表现我做过一个简单的压力测试一个包含50个步骤的任务链每个步骤执行一个sleep 0.1的命令。总执行时间大约是5.2秒其中引擎自身的开销配置解析、状态更新、日志写入大约占0.3秒。也就是说引擎开销不到总时间的6%对于大多数场景来说完全可以接受。如果步骤数量增加到500个引擎开销会线性增长但主要瓶颈在文件I/O上——每次状态更新都要读写status.json。如果你真的需要跑几百个步骤建议把状态存储换成SQLite或者减少状态更新的频率比如只在步骤开始和结束时更新而不是每次重试都更新。我在实际使用中单个任务链的步骤数通常不超过20个所以文件存储完全够用。如果你有更大规模的需求可以考虑把state.py里的存储后端抽象成接口方便替换。7. 一些个人体会和后续可以折腾的方向这套东西我从最初的一个bash脚本慢慢迭代到现在这个版本前后大概花了两个周末。最大的感受是自动化工具的价值不在于功能多而在于可靠。一个能稳定跑三个月不出问题的简单执行器比一个功能丰富但每周都要修的工具强得多。如果你打算自己动手我的建议是先从最核心的“顺序执行失败重试”开始跑通一个真实场景后再逐步加功能。不要一开始就想着支持并行、支持分布式、支持可视化——那些都是后面的事。先把一个步骤能可靠地跑起来再考虑两个步骤怎么串起来。后续我可能会尝试的方向有两个一是把状态存储换成SQLite支持更复杂的查询和统计二是增加一个简单的Web界面用来查看运行历史和触发任务。但这两个都不是必须的取决于实际需求。最后分享一个小技巧在配置里给每个步骤加一个description字段写清楚这一步在做什么。过几个月回头看的时候你会感谢自己当初写了注释。这个字段不影响执行但能大大降低维护成本。steps: - name: fetch description: 从上游接口拉取当日增量数据输出到 raw/ 目录 command: python fetch.py --date {{today}}好了关于rea的分享就到这里。如果你也在做类似的事情或者有更好的思路欢迎一起交流。工具是死的场景是活的适合自己的才是最好的。