Pipeline Starter

受众：开发者摘要：管道编排 Facade。Handler 分组、配置化启停、条件执行、next/skip/stop 中断、异常策略、步骤 trace 和执行监听。

何时使用

场景	建议
一段业务流程拆成多个有顺序步骤（校验链 / 前置检查）	`Pipeline.execute(name, context)`
多 Handler 可独立增删	Handler 通过 `@Pipeline.Step` 或 `getPipeline()` 绑定，按配置 order、注解 order、`order()`、Spring `Ordered` / `@Order` 排序
步骤需要查看执行轨迹和耗时	`trace-enabled: true`，按需接入 `PipelineExecutionListener`
不确定该用 Hook、Pipeline 还是 Chain	先看钩子 / 管道 / 责任链专题

Pipeline 只负责同一个上下文里的有序步骤执行。钩子 / 管道 / 责任链的选型、推荐场景、注解糖和组合使用见钩子 / 管道 / 责任链专题。

Starter 入口与源码骨架

Pipeline starter 采用 Facade -> Manager -> Registry / Executor -> Handler 的薄分层。业务代码默认使用 Pipeline.execute(...)；复杂装配、测试替换或按名称注入时使用 PipelineFactory 创建 PipelineFlow，再调用 pipeline.execute(ctx)。执行内核负责选择步骤、排序、trace、监听、异常策略和补偿。

层	类	职责
Facade	`Pipeline`	业务调用入口；支持 `@Pipeline.Execute` 自动触发、`@Pipeline.Step` / `@Pipeline.Action` 注册步骤、正式 Handler 和 `Pipeline.step(...)` / `Pipeline.action(...)` inline lambda
Flow	`PipelineFactory` / `PipelineFlow`	可注入实例入口：`pipelineFactory.flow(name).execute(context)`
Manager	`DefaultPipelineManager`	校验 Pipeline 名称和上下文，处理 Pipeline 级启停，委托 Registry / Executor，并在最终结果处通知监听器
Registry	`DefaultPipelineHandlerRegistry`	按 Pipeline 名绑定步骤，应用 Handler 启停配置，并按配置 order、注解 order、`PipelineHandler.order()`、Spring `Ordered` / `@Order` 排序
Executor	`DefaultPipelineExecutor`	同步执行步骤，解释 `next/skip/stop`，记录 trace，触发阶段监听，按需执行逆序补偿
Handler	`PipelineHandler` / `CompensatingPipelineHandler`	单个稳定步骤；返回 `next` 继续、`skip` 跳过、`stop` 正常中断
Listener	`PipelineExecutionListener`	承接审计、指标、执行日志和运维观测；监听器异常只记录 warning，不影响主流程
Context	`PipelineContext` / `DefaultPipelineContext` / `ExecutionKey<T>`	同一条链路内共享数据；`PipelineContext` 继承共享 `ExecutionContext`，稳定业务字段优先强类型 Context，临时数据用 `ExecutionKey<T>` 做类型校验

关键机制顺序：

Pipeline.execute(name, context) 调用 PipelineManager。
DefaultPipelineManager 校验参数和启停配置，向 Registry 取当前 Pipeline 的 Handler。
DefaultPipelineHandlerRegistry 完成绑定、启停过滤和排序。
DefaultPipelineExecutor 按顺序执行 Handler，supports=false 记为 skipped，handle=null 视为 next()。
stop() 是正常业务中断，会返回 INTERRUPTED 并触发已成功步骤的逆序补偿；异常按配置选择抛出或返回 FAILED。
PipelineExecutionListener 接收 start / beforeStep / afterStep / stop / error / compensate / final execution 事件。

边界

Pipeline 是进程内、同步、轻量、有序的业务步骤编排。它不内置异步、延迟、重试、持久化状态或分布式事务；业务最终状态、资金扣减、外部副作用和事务边界仍由领域 Service 控制。

生产接入样板

当前生产代码中的 Pipeline 接入点都保持短同步前置检查，Service 继续持有事务和最终副作用：

模块	Pipeline	说明
Forum	`forum.topic.publish.precheck` / `forum.reply.publish.precheck`	主题 / 回复发布前置规则校验，落库、计数和事件仍由 Forum Service 控制
Forum	`forum.topic.report.precheck` / `forum.reply.report.precheck` / `forum.report.status.update.precheck`	举报创建和举报状态处理前置校验，举报记录、处理记录和 Outbox 仍由 Forum Service 控制
CMS	`cms.content.review.submit.precheck` / `cms.content.review.finish.precheck`	内容提交审核和审核完成前置校验，审批工作流、状态字段和操作日志仍由 CMS Service 控制
Blog	`blog.comment.save`	评论命令规范化、文章可见性、父评论归属和内容安全结果传递，评论落库和计数仍由 Blog Service 控制
Mall	`mall.order.preview`	订单预览命令、地址、购物车快照、商品项数量和币种一致性校验，价格计算、优惠试算、子订单拆分和只读预览返回仍由 Mall Service 控制
Mall	`mall.order.create.prepare`	普通下单命令、地址、购物车快照、只读库存和币种一致性校验，库存扣减、优惠券锁定、订单落库和购物车清理仍由 Mall Service 控制
Mall	`mall.order.pay.prepare`	普通订单支付前置可支付状态、金额 / 币种、支付场景和余额可用性校验，Payment Provider 调用、收银台订单、订单状态、优惠券 / 积分 / 秒杀确认和虚拟交付仍由 Mall Service 控制
Web3	`money.web3.swap.quote.preview`	Swap 报价阶段执行 Token 风险预检和 DEX 报价，订单快照、txHash 绑定、receipt 同步和链上副作用仍由 Web3 Service 控制

快速开始

yaml

spring:
  open:
    pipeline:
      enabled: true

复制模板：Context

先定义一个强类型上下文。把类名、字段和 pipeline 名改成你的业务名即可：

java

public class OrderCreatePipelineContext extends DefaultPipelineContext {
    private final Long userId;
    private final Long amount;

    public OrderCreatePipelineContext(String traceId, Long userId, Long amount) {
        super(traceId);
        this.userId = userId;
        this.amount = amount;
    }

    public Long getUserId() {
        return userId;
    }

    public Long getAmount() {
        return amount;
    }
}

复制模板：注解 Step 类

推荐用 @Pipeline.Step 标注 Handler 类。把 pipeline 名、code、类名和上下文字段改成你的业务名即可：

java

@Component
@Pipeline.Step(
        value = "mall.order.create",
        code = "amount-check",
        order = 10,
        description = "Check order amount before creation.")
public class OrderAmountCheckStep implements PipelineHandler<OrderCreatePipelineContext> {

    @Override
    public boolean supports(OrderCreatePipelineContext context) {
        return context.getAmount() != null;
    }

    @Override
    public PipelineStepResult handle(OrderCreatePipelineContext context) {
        if (context.getAmount() <= 0) {
            return PipelineStepResult.stop("amount is invalid");
        }
        return PipelineStepResult.next();
    }
}

@Pipeline.Step 标在 Handler 类上后，可以省掉 getPipeline()、getCode()、getDescription() 和 order() 样板；配置里的 pipelines.<name>.handlers.<code>.order 仍然可以覆盖注解排序。

复制模板：普通 Bean 方法

简单步骤也可以直接写在任意 Spring Bean 方法上，不需要单独创建 Handler 类：

java

@Component
public class OrderCreatePipelineSteps {

    @Pipeline.Step(value = "mall.order.create", code = "amount-check", order = 10)
    public PipelineStepResult checkAmount(OrderCreatePipelineContext context) {
        if (context.getAmount() <= 0) {
            return PipelineStepResult.stop("amount is invalid");
        }
        return PipelineStepResult.next();
    }

    @Pipeline.Action(value = "mall.order.create", code = "write-log", order = 20)
    public void writeLog(OrderCreatePipelineContext context) {
        log.info("create order traceId={}", context.getTraceId());
    }
}

注解	方法签名	语义
`@Pipeline.Step`	`PipelineStepResult method(C context)`	返回 `next` / `skip` / `stop`，拥有完整步骤结果控制权
`@Pipeline.Action`	`void method(C context)`	只执行副作用，完成后自动 `next()`

复制模板：Service 调用

显式 Facade 调用最清楚，适合主流程里需要读取执行结果、自行决定后续动作的场景：

java

public class OrderService {
    public void createOrder(Long userId, Long amount) {
        OrderCreatePipelineContext context =
                new OrderCreatePipelineContext("trace-" + userId, userId, amount);

        PipelineExecutionResult<OrderCreatePipelineContext> result =
                Pipeline.execute("mall.order.create", context);

        if (result.isInterrupted()) {
            throw new IllegalStateException(result.getMessage());
        }

        // Pipeline 只做前置步骤编排；写订单、扣库存、创建支付单仍由 Service 控制。
    }
}

复制模板：自动触发业务方法

如果某个业务方法固定需要先跑同一条 Pipeline，可以直接标 @Pipeline.Execute。方法参数里必须有一个 PipelineContext；Pipeline 成功或被配置禁用后才会进入目标方法，stop() 会抛出 PipelineInterruptedException 并阻断目标方法。

@Pipeline.Execute 走 Spring AOP 代理；请从其他 Spring Bean 调用该方法，同一个类内部 this.createOrder(...) 自调用不会触发注解。

java

@Service
public class OrderService {

    @Pipeline.Execute("mall.order.create")
    public Order createOrder(OrderCreatePipelineContext context) {
        // 能执行到这里，表示 mall.order.create Pipeline 已经通过。
        return doCreateOrder(context.getUserId(), context.getAmount());
    }
}

上下文参数不在第一个位置时，用 contextArg 指定下标：

java

@Service
public class OrderService {

    @Pipeline.Execute(value = "mall.order.create", contextArg = 1)
    public Order createOrder(String requestSource, OrderCreatePipelineContext context) {
        return doCreateOrder(context.getUserId(), context.getAmount());
    }
}

Facade API

java

// 按 Pipeline 名执行（自动收集 Handler Bean 和 @Pipeline.Step / @Pipeline.Action 方法）
Pipeline.execute(pipelineName, context);

// 传入临时 Handler 列表（不依赖自动装配）
Pipeline.execute(pipelineName, context, handlers);

// 传入 inline step（不创建 Handler 类）
Pipeline.execute(pipelineName, context,
        Pipeline.action("print", ctx -> log.info("pipeline {}", ctx.getTraceId())),
        Pipeline.step("validate", ctx -> {
            if (ctx.get("amount") == null) {
                return PipelineStepResult.stop("amount required");
            }
            return PipelineStepResult.next();
        }));

Inline Step

简单的一次性动作不需要单独创建 PipelineHandler 类，可直接用 Facade 传 lambda：

java

DefaultPipelineContext context = new DefaultPipelineContext()
        .put("amount", 100L);

Pipeline.execute("mall.order.create", context,
        Pipeline.action("print", ctx -> log.info("before order create")),
        Pipeline.step("amount-check", ctx -> {
            Long amount = ctx.get("amount");
            if (amount == null || amount <= 0) {
                return PipelineStepResult.stop("amount invalid");
            }
            return PipelineStepResult.next();
        }));

API	说明
`@Pipeline.Execute`	标注业务方法，方法执行前自动跑指定 Pipeline；中断或失败会阻断目标方法
`@Pipeline.Step`	标注 Handler 类或普通 Bean 方法，注册正式可配置步骤
`@Pipeline.Action`	标注普通 Bean 方法，注册自动 `next()` 的动作步骤
`Pipeline.action(code, action)`	执行 void lambda，自动返回 `next()`
`Pipeline.step(code, callback)`	执行返回 `PipelineStepResult` 的 lambda，可 `next` / `skip` / `stop`
`.when(predicate)`	当前上下文满足条件时才执行该 inline step
`.named(name)` / `.described(description)`	覆盖 trace 中展示的名称和说明

Inline step 默认按传参顺序执行；如果配置了 pipelines.<name>.handlers.<code>.order，仍可通过配置覆盖顺序。正式、可复用、需要独立测试或配置启停的步骤仍建议沉淀为 PipelineHandler Bean。

Handler 返回值

返回值	说明
`PipelineStepResult.next()`	当前步骤完成，继续执行后续 Handler
`PipelineStepResult.skip()`	当前步骤跳过，继续执行后续 Handler
`PipelineStepResult.stop()` / `stop(message)`	正常中断流程，不视为异常；执行器会同步写入 `context.terminate(message)`，方便钩子 / 管道 / 责任链组合链路用 `ctx.isTerminated()` 统一判断

Saga 补偿（步骤回滚）

当链因某步骤 stop() 或抛出异常而中断时，执行器会对此前已成功执行且实现了 CompensatingPipelineHandler 的步骤按执行逆序依次调用 compensate(context)，用于撤销已写入的副作用（如已扣库存、已建记录）。

java

@Component
@Pipeline.Step(
        value = "mall.order.create",
        code = "reserve-stock",
        order = 20,
        description = "Reserve stock before order payment.")
public class ReserveStockStep implements CompensatingPipelineHandler<OrderCreatePipelineContext> {

    @Override
    public PipelineStepResult handle(OrderCreatePipelineContext context) {
        stockService.reserve(context.getSkuId(), context.getQuantity());
        return PipelineStepResult.next();
    }

    @Override
    public void compensate(OrderCreatePipelineContext context) {
        // 链后续步骤失败时，逆序回滚本步骤已写入的副作用
        stockService.release(context.getSkuId(), context.getQuantity());
    }
}

补偿规则：

规则	说明
补偿范围	仅补偿此前返回 `next()` 且实现了 `CompensatingPipelineHandler` 的步骤
不补偿	`supports()` 跳过的步骤、返回 `stop()` 的步骤自身、抛异常的步骤自身；只实现 `PipelineHandler`（无副作用）的步骤不进补偿栈、不产生 trace
补偿顺序	严格逆序：后执行的先补偿
补偿失败	由 `compensation-failure-strategy` 决定：`continue`（默认，记录后继续补下一个）/ `abort`（中止补偿）
trace	补偿步骤记 `COMPENSATED` / `COMPENSATION_FAILED` 轨迹
边界	编排式 Saga，适合单次调用内多步副作用回滚；跨服务 / 跨库强一致仍需 TCC / 消息事务，不是 Pipeline 职责

需要补偿的步骤实现 CompensatingPipelineHandler（extends PipelineHandler）即可；仍可用 @Pipeline.Step 绑定 Pipeline 名、code 和排序。只实现 PipelineHandler 的步骤视为无副作用、不参与补偿。补偿默认开启，可用 compensation-enabled: false 关闭。

执行结果

PipelineExecutionResult 包含：

字段	说明
`success`	是否执行成功
`status`	`SUCCESS` / `INTERRUPTED` / `FAILED` / `DISABLED`
`traceId`	当前上下文 traceId
`interrupted`	是否被 Handler 正常中断
`disabled`	该 Pipeline 是否被配置关闭
`interruptedBy`	中断或失败的 Handler
`message`	中断原因或异常消息
`exception`	异常对象（仅 `exception-strategy: result` 时）
`elapsedMillis`	总耗时
`traces`	每个 Handler 的执行轨迹

配置项

yaml

spring:
  open:
    pipeline:
      enabled: true
      exception-strategy: throw          # throw / result
      compensation-enabled: true
      compensation-failure-strategy: continue   # continue / abort
      trace-enabled: true
      execution-log-enabled: false
      pipelines:
        mall.order.create:
          enabled: true
          handlers:
            amount-check:
              enabled: true
              order: 10
            risk-check:
              enabled: false
        web3.tx.confirm:
          enabled: true

配置	默认值	说明
`enabled`	`true`	启用开关
`exception-strategy`	`throw`	Handler 异常时的处理（`throw` 直接抛出 / `result` 转为失败结果）
`compensation-enabled`	`true`	链中断 / 异常时是否对已执行且声明补偿的步骤逆序调用 `compensate()`
`compensation-failure-strategy`	`continue`	某步骤 `compensate()` 自身抛异常时：`continue` 继续补偿 / `abort` 中止
`trace-enabled`	`true`	是否记录步骤轨迹和耗时
`execution-log-enabled`	`false`	是否启用默认 JDK 日志监听器输出执行摘要
`pipelines.<name>.enabled`	`true`	指定 Pipeline 是否启用
`pipelines.<name>.handlers.<code>.enabled`	`true`	指定 Handler 是否启用
`pipelines.<name>.handlers.<code>.order`	空	覆盖 Handler 排序；为空时使用注解 order / `order()` / `Ordered` / `@Order`

Provider

不提供 Provider 切换。扩展能力通过 SPI：

实现 PipelineHandler<T> 并注册为 Spring Bean，推荐用 @Pipeline.Step 绑定 Pipeline 名和 code
通过注解 code 或 getCode() 提供稳定标识，配置化启停、排序覆盖和执行轨迹都使用该值
通过 supports(context) 做轻量条件执行，返回 false 时记录 SKIPPED 并继续后续步骤
通过注解 order、order()、Ordered 或 @Order 控制执行顺序；不设置时默认 Ordered.LOWEST_PRECEDENCE
有副作用、需要回滚的步骤改实现 CompensatingPipelineHandler<T>（extends PipelineHandler<T>），见 Saga 补偿章节
通过自定义 PipelineContext 子类承载领域数据，或用 ExecutionKey<T> 做类型安全的步骤间数据传递（见上下文章节）

执行记录

Pipeline 不直接依赖数据库，也不内置后台页面。生产环境需要落库、指标或审计时，实现监听器即可：

java

@Component
public class AuditPipelineExecutionListener implements PipelineExecutionListener {
    @Override
    public void onExecution(PipelineExecutionResult<? extends PipelineContext> result) {
        PipelineExecutionRecord record = PipelineExecutionRecord.from(result);
        // 保存 record 到业务审计、指标系统或数据库
    }
}

默认 JDK 日志监听器可按需开启：

yaml

spring:
  open:
    pipeline:
      execution-log-enabled: true

执行记录保留 pipeline、traceId、status、interruptedBy、message、异常类型、总耗时和步骤 trace。默认 throw 策略下，Handler 异常仍会向调用方抛出，但 PipelineManager 会先通知监听器收到 FAILED 快照。需要数据库持久化时，由业务模块或应用装配层提供自己的 PipelineExecutionListener，不要让 starter 反向依赖运行时核心或数据库业务表。

阶段钩子

PipelineExecutionListener 除最终 onExecution 外，还提供可选的阶段钩子，用于细粒度执行日志、耗时统计和审计：

钩子	触发时机
`onStart`	Pipeline 开始（第一个步骤前）
`beforeStep`	每个步骤执行前（含被 supports 跳过的步骤）
`afterStep`	每个步骤执行后，携带 next / skip / stop 结果
`onStop`	某步骤 `stop()` 导致中断
`onError`	某步骤抛出异常
`onCompensate`	某步骤被补偿，携带补偿是否成功
`onExecution`	Pipeline 结束，携带最终结果

阶段钩子全部为默认空方法，按需覆盖。监听器自身抛出的异常会被执行器或管理器隔离（记录但不传播），不影响 Pipeline 主流程。

上下文

简单流程直接用 DefaultPipelineContext：

java

DefaultPipelineContext context = new DefaultPipelineContext()
    .put("amount", 100)
    .put("currency", "CNY");

正式业务优先定义自己的强类型上下文，模板见上文 OrderCreatePipelineContext。上下文字段应表达业务输入、步骤间共享状态和 traceId，不要把完整业务 Service 塞进 context。

PipelineContext 继承 spring-open-starter-execution 的 ExecutionContext，因此 Pipeline 与 Hook / Chain 可以共享同一套 trace、attributes、类型安全 ExecutionKey<T> 和中止状态语义。Pipeline 不再维护专属 key 类型；步骤间临时数据和跨钩子 / 管道 / 责任链组合传递的数据都使用 ExecutionKey<T>。

类型安全数据传递（ExecutionKey）

需要在步骤间共享临时数据又不想定义子类字段时，用 ExecutionKey<T> 代替裸字符串 key。生产业务推荐使用 ExecutionKey.of(name, Class<T>)，取值时编译期即为目标类型，并在运行期校验同名 key 的值类型：

java

ExecutionKey<Long> AMOUNT = ExecutionKey.of("amount", Long.class);

context.set(AMOUNT, 100L);
Long amount = context.get(AMOUNT);   // 无需强转

ExecutionKey 与字符串 key 共用同一份 attributes 存储，可混用；适合简单 / inline 场景。承载完整领域数据仍优先用强类型上下文子类。

实例入口

需要把 Pipeline 按名称注入、便于测试替换或和 Hook / Chain 组合时，使用 PipelineFactory：

java

@RequiredArgsConstructor
public class OrderService {

    private final PipelineFactory pipelineFactory;

    public void createOrder(OrderCreatePipelineContext context) {
        PipelineFlow<OrderCreatePipelineContext> pipeline = pipelineFactory.flow("mall.order.create");
        pipeline.execute(context);
    }
}

异常策略

默认 throw：Handler 抛出异常时直接向外抛出，适合开发阶段和核心业务流程。

切换 result：把异常转换为失败执行结果（适合自定义降级 / 回滚链）：

yaml

spring:
  open:
    pipeline:
      exception-strategy: result

命名约定

Pipeline 名使用点分层：payment.create / web3.tx.confirm / ai-task.process
Handler code 使用短横线：amount-check / risk-check / create-user
Handler 命名以业务动作结尾：PaymentAmountCheckHandler / RiskCheckHandler
Handler 内只做单个步骤，不塞完整业务流程

注意事项

不在 Pipeline 中直接做长耗时阻塞任务；需要异步用 Queue 或 Scheduler 触发
Pipeline 不是工作流引擎、不做分布式事务、不替代队列或调度
中断用 PipelineStepResult.stop(...)；异常仅用于不可预期错误
Handler 应幂等（同一 context 多次执行结果一致），便于失败重试

后续方向

后续可按需扩展：

异步 Pipeline
与 Queue / Scheduler / Event 协作样板增强
Forum 内容发布前置检查真实接入
AI 任务处理管道

验证命令

bash

./mvnw -pl spring-open-starters/spring-open-starter-pipeline -am test -DskipITs -DskipFrontend

Pipeline Starter ​

何时使用 ​

Starter 入口与源码骨架 ​

边界 ​

生产接入样板 ​

快速开始 ​

复制模板：Context ​

复制模板：注解 Step 类 ​

复制模板：普通 Bean 方法 ​

复制模板：Service 调用 ​

复制模板：自动触发业务方法 ​

Facade API ​

Inline Step ​

Handler 返回值 ​

Saga 补偿（步骤回滚） ​

执行结果 ​

配置项 ​

Provider ​

执行记录 ​

阶段钩子 ​

上下文 ​

类型安全数据传递（ExecutionKey） ​

实例入口 ​

异常策略 ​

命名约定 ​

注意事项 ​

后续方向 ​

验证命令 ​

相关 ​

Pipeline Starter

何时使用

Starter 入口与源码骨架

边界

生产接入样板

快速开始

复制模板：Context

复制模板：注解 Step 类

复制模板：普通 Bean 方法

复制模板：Service 调用

复制模板：自动触发业务方法

Facade API

Inline Step

Handler 返回值

Saga 补偿（步骤回滚）

执行结果

配置项

Provider

执行记录

阶段钩子

上下文

类型安全数据传递（ExecutionKey）

实例入口

异常策略

命名约定

注意事项

后续方向

验证命令

相关