023、Workflow 编排实战：pipeline/parallel 的选择与 Barrier 机制

023、Workflow 编排实战：pipeline/parallel 的选择与 Barrier 机制

上周五凌晨三点，我盯着终端里那行血红的报错发呆——Claude Code 的 workflow 在并行执行到第 47 个任务时，突然把所有子进程的 stdout 混成了一锅粥。日志里task_47的输出里夹着task_12的报错信息，task_33的进度条直接覆盖了task_08的 JSON 输出。那一刻我意识到，workflow 编排不是简单的“把任务串起来”或者“一股脑并行”，pipeline 和 parallel 的选择背后藏着对资源、依赖和容错能力的深刻理解。

那个让我通宵的并行陷阱

先说说那个 bug 的根因。我当时的 workflow 长这样：

workflow:stages:-name:data_ingestionparallel:-task:fetch_from_s3-task:query_redshift-task:call_external_api-name:data_transformpipeline:-task:clean_data-task:feature_engineering-task:normalize

看起来没问题对吧？数据摄入阶段三个任务互不依赖，并行执行；转换阶段严格串行。但问题出在fetch_from_s3和query_redshift都往同一个临时目录写中间文件，而call_external_api又依赖那个目录里的某个锁文件——这他娘的是隐式依赖，workflow 引擎根本不知道。

并行执行时，三个任务同时启动，fetch_from_s3还没写完文件，call_external_api就开始读，读到的是一半的数据。更坑的是，query_redshift也在写同一个目录，文件名冲突直接覆盖了fetch_from_s3的输出。日志里看到的是task_47的报错，实际上根在task_12。

别这样写：把有隐式依赖的任务放在 parallel 块里。workflow 引擎只看显式依赖，你代码里的文件锁、共享变量、全局状态，它一概不知。

pipeline 和 parallel 的本质差异

很多人以为 pipeline 就是“串行”，parallel 就是“并行”，太肤浅了。从工程角度看，这两个模式对应着完全不同的资源模型和错误传播策略。

pipeline：顺序执行 + 状态传递

pipeline 的核心价值不是“慢”，而是确定性。每个任务执行时，前一个任务的结果已经稳定落盘，你可以放心地依赖上一个任务的输出。Claude Code 的 pipeline 实现里，每个 task 的 stdout/stderr 会被隔离到独立缓冲区，任务间通过显式的outputs字段传递数据。

pipeline:-task:extract_entitiesoutputs:entities:/tmp/entities.json-task:classify_entitiesinputs:entities:/tmp/entities.json# 这里踩过坑：如果 extract_entities 输出的是 JSON 数组，# classify_entities 必须用 json.load() 读，别用字符串拼接

pipeline 的容错策略是“一票否决”——任何一个任务失败，整个 pipeline 终止。这在数据清洗场景下是对的：脏数据进入下游只会制造更多脏数据。

parallel：并发执行 + 独立失败域

parallel 的核心价值是吞吐，但代价是状态隔离。每个并行任务应该是一个独立的、无副作用的单元。Claude Code 的 parallel 实现会为每个 task 分配独立的进程或容器，但共享文件系统——这就是上面那个 bug 的根源。

parallel:-task:process_chunk_1# 别这样写：所有 chunk 写同一个 output.json# 应该写：process_chunk_1 -> /tmp/chunk_1.json-task:process_chunk_2# process_chunk_2 -> /tmp/chunk_2.json

parallel 的容错策略更灵活：你可以设置max_failures参数，允许一定比例的任务失败而不终止整个 workflow。这在处理海量数据分片时特别有用——某个分片数据损坏，跳过它继续处理其他分片。

Barrier 机制：并行和串行的桥梁

真正让 workflow 编排变得复杂的是混合模式。你不可能永远只串行或只并行，真实场景往往是：并行处理一批数据，然后串行聚合结果，再并行分发下一批。

Barrier 就是干这个的。它像一个同步点，确保所有并行任务到达这个点之后，才能继续往下走。

workflow:stages:-name:parallel_processingparallel:-task:chunk_1-task:chunk_2-task:chunk_3-name:barriertype:sync# 这里踩过坑：barrier 必须等待所有 parallel 任务完成，# 包括那些已经失败的任务（如果 max_failures 允许）-name:aggregationpipeline:-task:merge_results-task:generate_report

Claude Code 的 barrier 实现有个隐藏行为：它会等待所有子任务完成，包括那些已经失败的任务。这意味着如果你的 parallel 块里有一个任务挂了，barrier 会一直等到那个任务的超时时间到达，然后才把失败状态传播到下游。别这样写：给 parallel 任务设置过长的超时时间，barrier 会变成性能瓶颈。

Barrier 的三种模式

根据我的踩坑经验，barrier 有三种实用模式：

1. 严格屏障（默认）：所有任务必须成功完成，否则 workflow 失败。适用于数据一致性要求高的场景，比如金融交易的对账。

2. 宽松屏障（max_failures > 0）：允许部分任务失败，但 barrier 仍然放行。适用于数据采集场景，某个数据源挂了不影响整体流程。

3. 超时屏障（timeout + partial）：等待指定时间后，未完成的任务被标记为“超时”，barrier 带着已完成的结果继续。适用于实时性要求高的场景，比如监控告警。

-name:data_collectionparallel:-task:source_a-task:source_b-task:source_cbarrier:mode:partialmax_failures:1timeout:30s# 这里踩过坑：timeout 必须大于最慢任务的预期执行时间，# 否则 barrier 会过早放行，下游拿到不完整的数据

实战中的选择策略

说了这么多理论，到底什么时候用 pipeline，什么时候用 parallel？我总结了一套自己的判断逻辑：

用 pipeline 的场景：

• 任务间有数据依赖，后一个任务必须用前一个任务的输出
• 错误传播需要严格阻断，脏数据不能流到下游
• 调试阶段，串行执行更容易定位问题
• 资源受限，无法同时运行多个任务（比如单 GPU 的模型推理）

用 parallel 的场景：

• 任务间无依赖，或者依赖可以通过外部存储解耦
• 需要提高吞吐量，且任务执行时间远大于调度开销
• 任务可以独立失败，不影响整体流程
• 数据分片处理，每个分片逻辑相同但数据不同

用 barrier 的场景：

• 并行阶段和串行阶段需要切换
• 需要控制并行任务的完成条件（全部成功、部分成功、超时）
• 需要收集并行任务的结果进行聚合

一个真实的混合编排案例

上个月我处理一个日志分析 workflow，数据量每天 500GB，需要分片处理然后聚合。最终方案是这样的：

workflow:stages:-name:shardpipeline:-task:split_logs# 按时间戳分片，输出 100 个文件-name:processparallel:-task:analyze_shard_1-task:analyze_shard_2# ... 100 个并行任务barrier:mode:partialmax_failures:5timeout:600s-name:mergepipeline:-task:collect_results# 这里踩过坑：collect_results 必须处理部分失败的情况，# 如果某个 shard 失败了，它的结果文件不存在，要跳过-task:generate_report

这个方案的关键在于 barrier 的max_failures: 5。实际运行中，确实有 2-3 个 shard 因为数据格式异常处理失败，但 barrier 放行了，merge 阶段跳过了这些 shard 的结果，最终报告仍然生成了。如果当时用了严格屏障，整个 workflow 就会因为几个异常 shard 而失败，需要人工介入重跑。

个人经验总结

1. 永远假设并行任务之间有隐式依赖。即使代码里没有显式依赖，文件系统、环境变量、全局锁都可能成为隐式依赖。解决方案：每个并行任务使用独立的临时目录，用 UUID 命名。

2. Barrier 的超时时间要留余量。我一般设置为最慢任务预期时间的 1.5 倍。太短会导致过早放行，太长会拖慢整体流程。

3. Pipeline 的失败传播要显式处理。Claude Code 的 pipeline 默认是“失败即终止”，但有时候你需要“失败后继续执行后续任务，但标记状态”。可以用on_failure: continue参数。

4. 日志隔离是并行调试的关键。每个并行任务的日志应该包含任务 ID 和时间戳，方便事后回溯。Claude Code 的--log-format json模式可以帮你做到这一点。

5. 不要迷信并行。并行不是银弹，调度开销、资源竞争、调试复杂度都是成本。如果一个任务执行时间小于 1 秒，串行执行可能比并行更快。

最后说一句：workflow 编排的本质不是“让任务跑起来”，而是“让任务在可控的约束下跑完”。pipeline 给你确定性，parallel 给你吞吐量，barrier 给你控制权。三者的组合，才是工程化的精髓。