一篇稿件进来：敏感词过滤、摘要抽取、封面裁剪、标签打点、索引写入。写成一个大方法，第一个需求变更就打穿它；拆成"阶段"又遇到新问题——阶段之间传什么、失败在哪停、耗时怎么算清。
管道的价值不在"分阶段"——任何方法拆成私有方法都算分阶段。价值在阶段契约足够窄，窄到可以单独替换、单独计时、单独重跑。下面按"契约 → 数据形态 → 失败 → 并行 → 观测 → 边界"拆。
1. 管道与责任链的差别
两条最实在的差别：责任链问"谁来处理"，管道问"数据到哪一步"；责任链环节接口统一（handle(ctx)），管道阶段可异构 Stage<I,O>。
| 维度 | 责任链 | 管道 |
|---|---|---|
| 语义 | 依次尝试，命中即停 | 依次变换，逐步推进 |
| 接口 | 统一 handle(Context) |
可异构 Stage<I,O> |
| 中断 | 常态（拒绝就停） | 异常态（失败才停） |
| 数据 | 共享同一个上下文 | 每阶段产生新值 |
| 失败处理 | 通常立即返回拒绝原因 | 需要显式的重试/降级/死信策略 |
| 顺序 | 可以靠 @Order 猜出意图 |
顺序即数据依赖，必须写死或声明 |
如果你发现自己写的"管道"每个阶段都返回同一个 Result 并且一到失败就返回，那它其实是责任链，别硬套 Stage<I,O>——多出来的类型参数只会增加噪音。
2. 阶段契约：Stage<I,O> 写成什么样
public interface Stage<I, O> {
O apply(I input) throws StageException;
default String name() { return getClass().getSimpleName(); }
}
// 顺序组合：任一段抛异常即整体失败
public final class Pipeline<T> {
private final List<Stage<T, T>> stages; // 同构串联时才好这么写
public T run(T in) throws StageException { /* 逐段 apply */ }
}
三个契约细节，缺一个半年后必咬人：
| 契约点 | 要求 | 不要求的后果 |
|---|---|---|
| 输入是否可为 null | 明确写 @Nullable 或 Objects.requireNonNull |
半路 NPE，栈里看不出是哪段的锅 |
| 是否允许原地修改输入 | 声明"纯函数，不改入参" | 上游复用同一对象时数据被下游篡改 |
| 异常类型 | 只抛 StageException（受检或运行时统一包装） |
上层 catch 写不准，只能 catch (Exception) |
代价：Java 没有高阶类型，串 5 个阶段要写一堆类型参数，编译器能查错但读起来累。阶段超过 4 个时改上下文对象更现实。
3. 上下文对象 vs 逐步变换
// 形态 A：逐步变换，类型安全
byte[] raw = fetch(url);
String text = extractor.apply(raw);
Document doc = parser.apply(text);
int score = scorer.apply(doc);
// 形态 B：上下文对象，阶段各自读写
public class ContentContext {
byte[] raw; String text; Document doc; int score;
Map<String, Object> attrs = new HashMap<>();
List<String> errors = new ArrayList<>();
}
public interface ContentStage {
void run(ContentContext ctx) throws StageException;
}
| 维度 | 逐步变换 | 上下文对象 |
|---|---|---|
| 类型安全 | 编译期强约束 | 弱，attrs 里全是 Object |
| 阶段可插拔 | 加减阶段要改类型签名 | 加减阶段不动签名 |
| 调试成本 | 低，断点看局部变量 | 高，得知道"这个字段是谁写的" |
| 上下文膨胀 | 不会 | 极易变成"上帝对象" |
实操取舍：阶段 ≤ 4 且数据形态线性演进 → 逐步变换；阶段 ≥ 5、有分支跳过、有可选阶段 → 上下文对象，但要加约束：阶段产物用 record 承载而不是 Map<String,Object>；每个可变字段标"唯一写入者"（// written by: ParserStage only）。没有这条，半年后没人知道 text 被谁改过。
4. 失败与重试放在哪一层
三层可选，选错就是重复数据的源头：
// 1) 阶段内重试（只适合幂等的远程调用）
public Doc apply(String text) throws StageException {
return Retryer.of(3, Duration.ofMillis(200)).run(() -> llm.parse(text));
}
// 2) 阶段间重试（整段重跑，要求阶段纯函数）
for (int i = 0; i < 2; i++) {
try { doc = parser.apply(text); break; }
catch (TransientException e) { if (i == 1) throw e; }
}
// 3) 整条管道重跑（要求全链路幂等 + 有去重键）
pipeline.run(ctx); // 入口用 traceId 做幂等锁
| 重试层级 | 适用 | 前提 | 坑 |
|---|---|---|---|
| 阶段内 | 单次远程调用超时、限流 | 该调用幂等 | 重试放大下游压力，必须带抖动 |
| 阶段间 | 阶段本身不稳定但纯 | 阶段无副作用 | 重跑会把下游已写入的数据写两遍 |
| 整条管道 | 消息消费失败重投 | 全链路幂等键 | 没有去重键就是数据重复制造机 |
一条硬规则：有副作用的阶段（写库、发消息、扣额度）不放在可重试段落里。如果业务要求"先写索引再回写状态"，把写索引拆成管道外的 outbox，管道只负责产出待写数据。
5. 并行阶段：fork-join 形态
摘要、标签、封面裁剪互不依赖，可以并行：
public Doc enrich(Doc doc) throws StageException {
CompletableFuture<String> summary = CompletableFuture.supplyAsync(() -> summarizer.apply(doc), pool);
CompletableFuture<List<String>> tags = CompletableFuture.supplyAsync(() -> tagger.apply(doc), pool);
try {
return doc.withSummary(summary.get(3, TimeUnit.SECONDS))
.withTags(tags.get(3, TimeUnit.SECONDS));
} catch (Exception e) {
summary.cancel(true); tags.cancel(true);
throw new StageException("enrich failed", e);
}
}
| 并行形态 | 适用 | 风险 |
|---|---|---|
CompletableFuture.allOf |
所有分支都必须成功 | 一个失败要主动 cancel 其余，否则泄漏任务 |
allOf + 超时 |
有 SLA 的在线链路 | 超时后子任务仍在跑，占线程 |
invokeAll + Future |
分支数量固定、需要逐个取结果 | 中断语义要自己处理 |
并行只在分支耗时接近且都超过毫秒级时划算，微秒级的纯计算并行化得不偿失。
6. 可观测性：每阶段耗时打点
不打点的话，线上慢了你只看到"接口 800ms"，无法归因。最小实现：
public <I, O> Stage<I, O> timed(Stage<I, O> stage, StageMetrics metrics) {
return input -> {
long start = System.nanoTime();
try {
O out = stage.apply(input);
metrics.record(stage.name(), "ok", System.nanoTime() - start);
return out;
} catch (RuntimeException e) {
metrics.record(stage.name(), "fail", System.nanoTime() - start);
throw e;
}
};
}
| 指标 | 维度 | 用来回答 |
|---|---|---|
| 阶段耗时 P50/P99 | stage |
哪一段是瓶颈 |
| 阶段失败率 | stage + error_type |
失败集中在哪 |
| 阶段输入/输出体积 | stage |
是不是某段把数据放大了 |
打点别用 System.currentTimeMillis()——同一毫秒内的两个阶段会得到 0ms，看起来"免费"。用 nanoTime。
7. 与事件驱动、工作流的边界
| 维度 | 管道 | 事件驱动 | 工作流引擎 |
|---|---|---|---|
| 方向 | 单向 | 扇出，多消费者 | 有向图，可分支可回退 |
| 状态 | 无状态，一次调用内完成 | 无中心状态，靠事件溯源 | 有持久化状态与补偿 |
| 失败 | 就地失败，调用方感知 | 靠重投 + 死信 | 补偿事务、人工干预 |
| 跨天/跨服务 | 不合适 | 合适 | 合适 |
| 调试 | 单机断点即可 | 需要链路追踪 | 需要流程实例视图 |
判断口径：一次请求内跑完、失败可整体重来 → 管道；要等回调、要人审批、要跨天 → 工作流。 用管道模拟"等审批"，最后一定写出一个轮询数据库的 while 循环。
8. 两个真实场景
8.1 内容处理管线
| 阶段 | 输入 → 输出 | 失败策略 | 可跳过 |
|---|---|---|---|
| Fetch | url → byte[] | 重试 2 次 | 否 |
| Extract | byte[] → text | 直接失败 | 否 |
| Clean | text → text | 降级为原文 | 是 |
| Summarize | text → summary | 超时降级为空 | 是 |
| Index | doc → docId | 失败进 outbox 重投 | 否 |
注意 Clean 和 Summarize 的差别：Clean 失败降级（原文照样能发），Summarize 超时降级（摘要为空也能发）。降级策略要写在阶段契约里，不是写在调用方，否则同一阶段在不同调用方行为不一致。
8.2 支付前置校验链
风控、限额、券、余额四项检查。它们表面像管道，实际更像责任链：任一失败即拒绝，且要返回具体原因。硬套管道会逼出"每段都返回 CheckResult 并传递"的写法，等于把责任链包了一层壳。识别方法：如果每个阶段的输出类型都一样，并且失败即终止，那就是责任链。
9. 常见坑
| 现象 | 原因 | 处理 |
|---|---|---|
| 上下文越来越大，字段几十个 | 阶段把中间态都塞进 ctx | 阶段产物用 record，ctx 只放跨阶段必需项 |
| 调换两个阶段顺序就出错 | 隐含数据依赖没声明 | 阶段声明 requires() / produces()，启动时校验 |
| 一个事务包住整条管道 | 远程调用在事务里，连接被长时间占用 | 事务只在单个阶段内，跨阶段用 outbox |
| 某阶段偶发失败但无法重跑 | 该阶段有副作用 | 拆出副作用，管道只产数据 |
| 并行化后线程池打满 | 共享业务线程池 | 管道用独立、有界的池；分支带超时 |
第三条值得展开：管道里塞 @Transactional 是自然冲动（"要么全成功"），但管道常含 HTTP 与文件 IO，事务开着等网络，连接池几分钟就枯。做法是阶段内本地事务 + 阶段间最终一致。
10. 边界：什么时候别用管道
| 场景 | 更合适 | 理由 |
|---|---|---|
| 阶段间要反复回退重算 | 递归/迭代求解 | 管道是单向的 |
| 需要在任意阶段插入人工干预 | 工作流 | 管道无持久化状态 |
| 每个阶段输出同构、失败即停 | 责任链 | 管道契约是多余的抽象 |
| 只有 2 个阶段 | 两个方法调用 | 引入 Stage 接口纯属自嗨 |
| 阶段间需要事务强一致 | 单个服务方法 + 本地事务 | 跨阶段事务在分布式下不成立 |
11. 自测
- 什么信号说明你写的"管道"其实是责任链？
Stage<I,O>契约里必须显式声明哪三件事？- 上下文对象在什么条件下开始变成上帝对象，怎么约束？
- 有副作用的阶段为什么不能放进可重试段落？
- 并行阶段里一个分支失败，剩下分支该怎么处理？
- 阶段耗时打点为什么不能用
currentTimeMillis？