rea 模式解析：定义、编排、执行三步构建可复用自动化流程

1. 从“rea”这个标题说起：一个被低估的缩写背后藏着什么

第一次看到“rea”这个标题的时候，我脑子里蹦出来的第一反应是——这大概率又是一个被缩写玩坏的项目名。做技术的人都有个毛病，喜欢把长名字砍成三四个字母，图省事，结果过两个月自己都忘了当初是啥意思。但“rea”这个组合有点意思，它不像“api”“sdk”那种一眼就能对上号的行业黑话，也不像随便敲键盘滚出来的随机串。我翻了一圈社区里的讨论，发现围绕这个词的解读五花八门：有人说是“reactive”的前三个字母，有人猜是“read-eval-apply”的循环缩写，还有人把它跟“real-time”绑在一起聊。

这种模糊性反而让我来了兴趣。一个标题能同时勾出这么多方向的联想，说明它踩中了某个真实存在的需求交叉点——大家手里都有一堆零散的工具、脚本、流程，缺的不是功能，而是一个能把它们串起来的轻量级抽象层。“rea”恰好可以成为这样一个符号：它不绑定具体技术栈，不承诺大而全的框架，只解决一个核心问题——让重复性的操作变得可描述、可复用、可组合。

我打算按这个思路往下拆。不管你是写业务代码的、做数据处理的、还是折腾自动化流程的，只要日常工作里存在“同样的动作反复做、每次都要重新拼一遍”的情况，“rea”这套思路就能直接抄作业。它不要求你换语言、换平台、换工具链，核心就三件事：定义动作、编排顺序、处理结果。下面我会把这三点掰开揉碎，配上我实际踩过的坑和验证过的参数，争取让你看完就能在自己的环境里跑起来。

2. 核心思路拆解：为什么是“定义-编排-执行”这三板斧

2.1 从“脚本堆叠”到“动作抽象”的必然性

我见过太多项目死在“脚本堆叠”上。一开始只是三行命令，后来加个判断，再加个循环，再套个异常捕获，半年后这个文件变成八百行，谁都不敢动。问题的根源不在于脚本本身，而在于没有把“做什么”和“怎么做”分开。你写curl的时候，脑子里想的是“我要拿这个接口的数据”，但代码里写的是具体的 URL、header、超时参数。下次换个接口，这些细节全要重写一遍。

“rea”的第一个字母 R，我倾向于把它理解为Represent——把动作表示出来。不是写死一段代码，而是定义一个“动作描述”。这个描述里只包含三样东西：动作的名字、它需要什么输入、它会产生什么输出。至于内部是用 shell 调、用 Python 写、还是发个 HTTP 请求，那是实现层的事，跟调用方无关。

举个例子。假设你每天都要从某个数据源拉一份报表，然后清洗、入库、发通知。传统写法是一个大脚本从头到尾。用“rea”的思路，你会拆成四个动作：fetch_report、clean_data、load_db、notify。每个动作单独定义，单独测试。哪天数据源换了，只改fetch_report的实现，其他三个动作原封不动。这就是抽象带来的隔离价值。

注意：动作的粒度很关键。太粗，复用性差；太细，编排复杂度爆炸。我的经验是，一个动作只做一件“能用一句话说清楚”的事。如果描述里出现了“并且”“然后”，就该考虑拆了。

2.2 编排层为什么不能省

有人会问：我定义了一堆动作，直接按顺序调用不就行了，要什么编排层？这个问题我当初也想过，直到遇到第一个真实场景——条件分支和错误恢复。顺序调用只能处理“一切顺利”的情况。一旦某个动作失败，你是重试、跳过、还是回滚？如果某个动作的输出要作为下一个动作的输入，这个数据怎么传？如果两个动作之间需要等待一段时间，或者需要并发执行，顺序调用根本表达不了。

编排层的价值就在这儿：它把“动作之间的依赖关系”变成一等公民。你可以用声明式的方式描述：A 完成后触发 B 和 C，B 成功后再执行 D，C 失败则执行补偿动作 E。这些逻辑如果散落在代码里，就是一堆if-else和try-catch；如果抽到编排层，就变成一张清晰的关系图。

我实测下来，编排层最实用的三个能力是：顺序控制、条件分支、错误重试。顺序控制不用多说，就是定义谁先谁后。条件分支允许你根据上一个动作的输出决定下一步走向。错误重试则是给每个动作配置重试次数和退避策略，避免因为网络抖动导致整个流程挂掉。

2.3 执行器的选型逻辑

“rea”的最后一个字母 A，我把它对应到Apply——执行。执行器是整个体系里最“脏”的部分，因为它要跟真实环境打交道：文件系统、网络、数据库、外部命令。选型的时候我踩过几个坑，这里直接给结论。

第一，不要自己造执行器。除非你的需求极其特殊，否则用现成的进程管理或任务队列工具就够了。自己写执行器意味着你要处理进程信号、超时杀死、输出捕获、环境隔离，这些细节能吃掉你 80% 的时间。

第二，执行环境要隔离。我早期图省事，所有动作都在同一个 shell 会话里跑，结果一个动作改了环境变量，后面的动作全受影响。后来改成每个动作独立子进程，问题消失。代价是启动开销变大，但对于大多数场景，这点开销完全可以接受。

第三，输出格式要统一。动作的返回值不能是五花八门的字符串，最好统一成结构化的数据（比如 JSON）。这样编排层才能可靠地解析和传递。我见过有人用echo输出纯文本，然后在下一个动作里用grep提取，稍微改个格式就全崩了。

3. 核心细节解析：动作定义、参数传递与状态管理

3.1 动作定义的四个必备字段

一个合格的动作定义，至少包含四个字段：名称、输入模式、输出模式、执行体。名称是唯一标识，用来在编排层引用。输入模式和输出模式是契约，告诉调用方“我需要什么、我给你什么”。执行体是具体实现，可以是一个命令、一段代码、或者一个远程调用。

输入模式和输出模式我强烈建议用 JSON Schema 来描述。虽然写起来有点啰嗦，但它带来的好处是巨大的：调用方可以提前校验参数，编排层可以自动生成文档，执行体内部也可以根据 schema 做默认值填充和类型转换。我试过用纯字符串描述参数，结果就是每个动作的实现里都要写一堆if not param: param = default的防御代码，重复且容易漏。

{ "name": "fetch_report", "input": { "type": "object", "properties": { "date": { "type": "string", "format": "date" }, "source": { "type": "string", "enum": ["a", "b", "c"] } }, "required": ["date"] }, "output": { "type": "object", "properties": { "rows": { "type": "array" }, "count": { "type": "integer" } } } }

上面这个例子定义了一个拉取报表的动作。输入要求一个日期和一个可选的来源，输出是一个包含行数据和计数的对象。有了这个定义，编排层就知道调用它的时候必须传date，而source可以省略。执行体内部拿到的一定是符合 schema 的数据，不需要再做类型检查。

提示：schema 的required字段要慎用。我一开始把所有参数都标成必填，结果编排的时候发现很多参数其实有合理的默认值。后来改成只有真正无法推断的才标必填，灵活性提升很多。

3.2 参数传递的三种模式与选择依据

动作之间的参数传递，我总结下来有三种模式：直接传递、上下文传递、外部存储传递。每种都有适用场景，选错了会带来不必要的复杂度。

直接传递最简单：上一个动作的输出直接作为下一个动作的输入。适合数据量小、结构简单的场景。比如一个动作返回{"id": 123}，下一个动作接收这个对象，取id字段用。缺点是如果中间隔了好几个动作，参数要一层层往下传，链路上任何一环改了结构，后面全要跟着改。

上下文传递是引入一个共享的上下文对象，所有动作都能读写。适合需要跨多个动作共享状态的场景。比如第一个动作获取了认证令牌，后面所有动作都要用。缺点是上下文变成了隐式的全局变量，调试的时候很难追踪某个值是谁写的。我的做法是给上下文加命名空间，每个动作只能写自己命名空间下的键，读的时候可以跨命名空间读。这样既保留了共享能力，又限制了污染范围。

外部存储传递是把中间结果写到文件或数据库，下一个动作根据约定的位置去读。适合数据量大或者需要持久化的场景。比如一个动作生成了几百兆的中间文件，通过上下文传递会撑爆内存，写到磁盘再读就稳妥得多。缺点是要管理存储的生命周期，用完要清理，否则磁盘会爆。

我实际项目里的组合策略是：小对象走直接传递，认证和配置走上下文，大文件走外部存储。三条通道各司其职，互不干扰。

3.3 状态管理：幂等性与断点续跑

状态管理是“rea”体系里最容易被忽视、但出事最多的部分。核心就两个问题：动作是否幂等、流程能否断点续跑。

幂等性指的是同一个动作执行多次，结果是否一致。拉取报表这种动作通常是幂等的，拉一次和拉十次结果一样。但“写入数据库”就不一定了，重复执行可能产生重复数据。对于非幂等的动作，编排层需要记录执行状态，避免重复触发。我的做法是给每个动作实例分配一个唯一 ID，执行前先查状态表，如果已经成功过就直接跳过。

断点续跑指的是流程中断后，能否从断点继续，而不是从头再来。这个能力在长流程里非常关键。想象一个流程有二十个动作，跑到第十九个失败了，如果没有断点续跑，你只能从头再跑一遍，前十八个动作的时间全白费。实现断点续跑的关键是持久化每个动作的执行结果和状态。我用的是轻量级的本地文件存储，每个动作完成后写一个状态文件，记录动作名、输入哈希、输出、时间戳。重启时读取这些文件，已经成功的动作直接跳过。

注意：断点续跑的前提是动作幂等，或者至少能安全地跳过。如果一个动作有副作用且不幂等，跳过它可能导致数据不一致。这种动作要么改造成幂等，要么在续跑时强制重新执行。

4. 实操过程：从零搭一个最小可用的“rea”流程

4.1 环境准备与目录结构

我不打算推荐任何特定的框架，因为“rea”本身是一种思路，用你手头最顺手的工具实现就行。下面我用一个通用的目录结构来演示，你可以直接套用到自己的项目里。

rea-demo/ ├── actions/ # 存放动作定义 │ ├── fetch.json │ ├── clean.json │ └── load.json ├── impl/ # 存放动作实现 │ ├── fetch.sh │ ├── clean.py │ └── load.sh ├── flows/ # 存放编排定义 │ └── daily.yaml ├── state/ # 存放执行状态 └── run.sh # 入口脚本

这个结构的好处是定义与实现分离。actions/里只放契约，impl/里放具体代码。换实现的时候，契约不动，调用方无感知。flows/里放编排逻辑，一个流程一个文件。state/是运行时生成的，不用手动维护。

环境依赖方面，我假设你有基础的 shell 环境和一个能跑 Python 的解释器。不需要额外安装什么重型框架。如果你用其他语言，把impl/里的脚本换成对应语言的即可，思路完全一样。

4.2 定义第一个动作：拉取数据

先写动作定义。actions/fetch.json：

{ "name": "fetch", "description": "从指定来源拉取原始数据", "input": { "type": "object", "properties": { "source": { "type": "string" }, "date": { "type": "string" } }, "required": ["source", "date"] }, "output": { "type": "object", "properties": { "raw_path": { "type": "string" }, "row_count": { "type": "integer" } } } }

这个动作接收来源和日期，输出原始文件的路径和行数。注意输出里的raw_path是一个文件路径，而不是数据本身。这是有意为之——原始数据可能很大，放在文件里比放在内存里传递更稳妥。

对应的实现impl/fetch.sh：

#!/bin/bash set -euo pipefail SOURCE="$1" DATE="$2" OUTPUT_DIR="./data/raw" mkdir -p "$OUTPUT_DIR" OUTPUT_FILE="$OUTPUT_DIR/${SOURCE}_${DATE}.csv" # 模拟拉取过程，实际替换成你的数据源命令 echo "id,name,value" > "$OUTPUT_FILE" echo "1,alpha,100" >> "$OUTPUT_FILE" echo "2,beta,200" >> "$OUTPUT_FILE" ROW_COUNT=$(tail -n +2 "$OUTPUT_FILE" | wc -l) # 输出结构化结果 cat <<EOF { "raw_path": "$OUTPUT_FILE", "row_count": $ROW_COUNT } EOF

这个脚本做了几件事：创建输出目录、生成模拟数据、统计行数、输出 JSON。set -euo pipefail是必须的，保证任何一步出错就立即退出，不会带着半成品继续往下跑。

提示：输出 JSON 的时候要小心特殊字符转义。如果路径里包含空格或引号，直接拼字符串会生成非法 JSON。生产环境建议用jq来构造输出，或者用 Python 的json.dumps。

4.3 定义清洗动作与编排文件

清洗动作的定义actions/clean.json：

{ "name": "clean", "description": "清洗原始数据，去除空值和重复行", "input": { "type": "object", "properties": { "raw_path": { "type": "string" } }, "required": ["raw_path"] }, "output": { "type": "object", "properties": { "clean_path": { "type": "string" }, "removed_count": { "type": "integer" } } } }

实现impl/clean.py：

import sys import json import csv def main(): input_data = json.loads(sys.argv[1]) raw_path = input_data["raw_path"] seen = set() rows = [] removed = 0 with open(raw_path, "r") as f: reader = csv.DictReader(f) for row in reader: key = tuple(sorted(row.items())) if key in seen or not all(row.values()): removed += 1 continue seen.add(key) rows.append(row) clean_path = raw_path.replace("/raw/", "/clean/") import os os.makedirs(os.path.dirname(clean_path), exist_ok=True) with open(clean_path, "w", newline="") as f: writer = csv.DictWriter(f, fieldnames=rows[0].keys()) writer.writeheader() writer.writerows(rows) print(json.dumps({ "clean_path": clean_path, "removed_count": removed })) if __name__ == "__main__": main()

这个脚本接收一个 JSON 字符串作为参数，解析出raw_path，然后做去重和空值过滤。输出同样是 JSON。注意它把清洗后的文件写到了clean/目录，跟原始数据分开存放，方便对比和回滚。

编排文件flows/daily.yaml：

name: daily_report steps: - action: fetch params: source: "system_a" date: "${today}" output_key: fetch_result - action: clean params: raw_path: "${fetch_result.raw_path}" output_key: clean_result retry: max_attempts: 3 delay_seconds: 5

这个编排定义了两个步骤。第一步调用fetch，参数里的${today}是内置变量，运行时替换成当天日期。输出保存到fetch_result。第二步调用clean，参数从fetch_result里取raw_path。如果失败，最多重试三次，每次间隔五秒。

4.4 执行入口与状态记录

入口脚本run.sh负责解析编排文件、按顺序执行动作、记录状态：

#!/bin/bash set -euo pipefail FLOW_FILE="$1" STATE_DIR="./state" mkdir -p "$STATE_DIR" # 这里用 Python 做编排解析，实际可以用任何语言 python3 - <<'PYEOF' import yaml import json import subprocess import os import hashlib from datetime import datetime with open(os.environ["FLOW_FILE"]) as f: flow = yaml.safe_load(f) context = {"today": datetime.now().strftime("%Y-%m-%d")} state_dir = os.environ["STATE_DIR"] for step in flow["steps"]: action = step["action"] params = {} for k, v in step.get("params", {}).items(): if isinstance(v, str) and v.startswith("${") and v.endswith("}"): expr = v[2:-1] if "." in expr: obj, attr = expr.split(".", 1) params[k] = context[obj][attr] else: params[k] = context.get(expr, v) else: params[k] = v # 计算输入哈希，用于断点续跑判断 input_hash = hashlib.md5( json.dumps(params, sort_keys=True).encode() ).hexdigest() state_file = os.path.join(state_dir, f"{action}_{input_hash}.json") if os.path.exists(state_file): with open(state_file) as f: result = json.load(f) print(f"[skip] {action} already done") else: print(f"[run] {action} with {params}") impl = f"./impl/{action}.sh" if not os.path.exists(impl): impl = f"./impl/{action}.py" if impl.endswith(".py"): cmd = ["python3", impl, json.dumps(params)] else: cmd = [impl] + [str(v) for v in params.values()] proc = subprocess.run(cmd, capture_output=True, text=True) if proc.returncode != 0: print(f"[error] {action} failed: {proc.stderr}") raise SystemExit(1) result = json.loads(proc.stdout) with open(state_file, "w") as f: json.dump(result, f) if "output_key" in step: context[step["output_key"]] = result print("[done] flow completed") PYEOF

这个脚本的核心逻辑是：遍历步骤、解析参数、检查状态文件、执行动作、保存结果。状态文件的命名用了动作名加输入哈希，保证同样的输入不会重复执行。如果状态文件存在，直接跳过，实现断点续跑。

运行方式：

export FLOW_FILE="./flows/daily.yaml" export STATE_DIR="./state" bash run.sh

第一次运行会依次执行fetch和clean，第二次运行会直接跳过，因为状态文件已经存在。如果你修改了参数，输入哈希变化，对应的动作会重新执行。

5. 常见问题与排查技巧实录

5.1 动作执行失败但错误信息不明确

这是最常见的问题。动作脚本报错，但编排层只看到“退出码非零”，不知道具体原因。我的排查步骤是：先看动作脚本的 stderr 有没有输出，如果没有，在脚本开头加set -x打开调试模式，把每一步命令打印出来。如果 stderr 有输出但被编排层吞了，检查subprocess.run的capture_output参数，确保 stderr 被捕获并打印。

还有一个隐蔽的坑：动作脚本里用了管道，管道中间的命令失败但整体退出码是零。比如cat file | grep pattern，如果file不存在，cat失败，但grep可能返回零，导致整体退出码为零。解决办法是加set -o pipefail，让管道中任何一个命令失败都导致整体失败。

5.2 参数传递时类型丢失

JSON 里的数字传到 shell 脚本里会变成字符串，再传回来可能变成字符串类型的数字。如果下游动作期望整数，就会出错。我的做法是在动作定义里明确类型，执行体内部做一次显式转换。比如 Python 里用int(input_data["count"])，shell 里用$((COUNT + 0))。不要依赖隐式转换，不同语言的行为不一致。

5.3 状态文件堆积导致磁盘满

断点续跑的状态文件会一直累积。如果流程每天跑一次，一年就是三百多个文件。单个文件不大，但数量多了也占空间。我的清理策略是：状态文件保留最近七天的，更早的自动删除。在入口脚本里加一段清理逻辑，按修改时间排序，删掉过期的。注意不要删正在使用的状态文件，清理前检查文件锁或者用时间阈值过滤。

5.4 并发执行时的状态冲突

如果同一个流程被同时触发两次，两个实例可能同时读写同一个状态文件，导致数据错乱。解决办法是给状态文件加锁，或者用实例 ID 隔离状态目录。我用的是后者：每次运行生成一个唯一的运行 ID，状态文件放在state/{run_id}/下。这样不同实例互不干扰，代价是断点续跑需要指定运行 ID，不能自动恢复。权衡下来，对于大多数场景，隔离比自动恢复更重要。

5.5 动作重试导致副作用重复

重试机制很实用，但如果动作不幂等，重试会产生重复副作用。比如“发送通知”这个动作，第一次执行成功了但返回超时，编排层以为失败就重试，结果用户收到两条通知。解决办法是给动作加幂等键，执行前先检查这个键是否已经处理过。幂等键可以是输入参数的哈希，也可以是业务上的唯一标识。执行体内部维护一个已处理键的集合，重复的键直接返回上次的结果。

注意：幂等键的存储要考虑持久化。如果存在内存里，进程重启就丢了。我一般用轻量级的本地数据库或者带过期时间的文件来存。

6. 扩展方向：从单机脚本到可观测的流程体系

6.1 给动作加上指标采集

最小可用版本跑通之后，下一步自然是加可观测性。每个动作执行时记录三个指标：执行时长、成功/失败、输出大小。这些指标可以写到本地文件，也可以推到监控系统。我习惯先用本地文件，格式用 JSON Lines，每行一条记录。这样既方便人看，也方便后续用脚本做聚合分析。

# 在动作执行前后记录时间戳 START=$(date +%s%N) # ... 执行动作 ... END=$(date +%s%N) DURATION=$(( (END - START) / 1000000 )) echo "{\"action\":\"$ACTION\",\"duration_ms\":$DURATION,\"status\":\"$STATUS\"}" >> metrics.jsonl

有了这些数据，你就能回答一些很实际的问题：哪个动作最慢、哪个动作失败率最高、流程的整体耗时趋势是什么。这些问题在流程简单的时候不重要，但一旦动作数量超过十个，没有指标就是盲人摸象。

6.2 编排文件的模块化

当流程变多之后，你会发现很多步骤是重复的。比如“拉取-清洗-入库”这个组合在多个流程里出现。这时候可以把编排也做成可复用的模块。一个模块就是一组动作的集合，对外暴露输入和输出。其他流程引用这个模块，就像引用一个普通动作一样。

模块化的关键是接口稳定。模块内部的实现可以随便改，但只要输入输出契约不变，调用方就不受影响。这跟动作定义的思路是一致的，只是抽象层级更高了一层。我实际用下来，模块化能把编排文件的体积减少一半以上，而且修改公共逻辑的时候只需要改一处。

6.3 从命令行到服务化

命令行版本适合个人使用和调试，但如果要让团队共用，就需要服务化。服务化的核心是把编排引擎做成一个常驻进程，接收 HTTP 请求触发流程，返回执行 ID，再通过另一个接口查询状态。这样不同的人、不同的系统都能触发同一个流程，而不需要登录到同一台机器上敲命令。

服务化之后，状态管理也要跟着升级。本地文件换成数据库，支持并发读写和事务。执行日志集中收集，方便排查。权限控制也要加上，不是所有人都能触发所有流程。这些改造工作量不小，但如果你发现流程已经被超过三个人使用，或者被其他系统调用，那就值得投入。

我个人在实际操作中的体会是，“rea”这套思路最大的价值不在于某个具体工具，而在于它强迫你把“做什么”和“怎么做”分开思考。一旦养成这个习惯，哪怕不用任何框架，只是用最朴素的脚本加一个约定俗成的目录结构，代码的可维护性也会有质的提升。最后再分享一个小技巧：动作定义文件里的description字段一定要认真写，不要敷衍。三个月后你回头看，唯一能帮你快速回忆起这个动作是干嘛的，就是那行描述。