C# 管道模式（Pipeline Pattern）详解

一、核心思想

你列出的六点本质上是责任链模式 + 组合模式 + 依赖注入的组合应用。核心就一句话：

把"流程"从一段又长又硬编码的方法，拆解成一组可插拔的 Step，由管道按顺序编排执行。

传统写法： 管道写法： DoEverything() Pipeline { ├── Step A stepA(); ├── Step B stepB(); ├── Step C stepC(); └── Step D }

二、为什么需要它？（对比痛点）

❌ 传统写法的痛点

publicclassOrderService{publicvoidPlaceOrder(Orderorder){// 1. 校验if(order.Items.Count==0)thrownewException("空订单");// 2. 计算价格order.Total=order.Items.Sum(i=>i.Price*i.Qty);// 3. 库存扣减_inventory.Deduct(order);// 4. 支付_payment.Charge(order);// 5. 发通知_email.Send(order);// 6. 记录日志_log.Info("下单成功");// 7. 后续还要加：风控、优惠券、积分...}}

问题：

• 一个方法越来越长，可读性差
• 加/删步骤要改这个方法，违反开闭原则
• 没法单独测"库存扣减"这一步，难测试
• 其他业务想复用"发通知"逻辑，只能复制粘贴，难复用

三、核心抽象

1. 上下文（Context）—— 在 Step 之间传递数据

publicclassOrderContext{publicOrderOrder{get;set;}=new();publicboolIsValid{get;set;}publicdecimalDiscount{get;set;}publicList<string>Logs{get;}=new();}

2. Step 接口 —— 每个步骤只做一件事

publicinterfaceIStep<TContext>{TaskExecuteAsync(TContextcontext,CancellationTokenct=default);}

如果步骤需要决定"是否继续往下走"，可以返回一个结果：

publicenumStepResult{Continue,Stop}publicinterfaceIStep<TContext>{Task<StepResult>ExecuteAsync(TContextcontext,CancellationTokenct=default);}

3. Pipeline —— 负责按顺序编排

publicinterfaceIPipeline<TContext>{TaskExecuteAsync(TContextcontext,CancellationTokenct=default);}publicclassPipeline<TContext>:IPipeline<TContext>{privatereadonlyIEnumerable<IStep<TContext>>_steps;publicPipeline(IEnumerable<IStep<TContext>>steps){_steps=steps;}publicasyncTaskExecuteAsync(TContextcontext,CancellationTokenct=default){foreach(varstepin_steps){varresult=awaitstep.ExecuteAsync(context,ct);if(result==StepResult.Stop)break;}}}

四、完整示例：订单处理管道

Step 1 — 校验

publicclassValidateOrderStep:IStep<OrderContext>{publicTask<StepResult>ExecuteAsync(OrderContextctx,CancellationTokenct=default){if(ctx.Order.Items.Count==0){ctx.IsValid=false;ctx.Logs.Add("订单为空，终止流程");returnTask.FromResult(StepResult.Stop);}ctx.IsValid=true;ctx.Logs.Add("校验通过");returnTask.FromResult(StepResult.Continue);}}

Step 2 — 计算价格

publicclassCalculatePriceStep:IStep<OrderContext>{publicTask<StepResult>ExecuteAsync(OrderContextctx,CancellationTokenct=default){ctx.Order.Total=ctx.Order.Items.Sum(i=>i.Price*i.Quantity)-ctx.Discount;ctx.Logs.Add($"价格计算完成:{ctx.Order.Total}");returnTask.FromResult(StepResult.Continue);}}

Step 3 — 扣库存

publicclassDeductInventoryStep:IStep<OrderContext>{privatereadonlyIInventoryService_inventory;publicDeductInventoryStep(IInventoryServiceinventory)=>_inventory=inventory;publicasyncTask<StepResult>ExecuteAsync(OrderContextctx,CancellationTokenct=default){varok=await_inventory.DeductAsync(ctx.Order,ct);if(!ok){ctx.Logs.Add("库存不足，终止流程");returnStepResult.Stop;}ctx.Logs.Add("库存扣减成功");returnStepResult.Continue;}}

Step 4 — 支付

publicclassPaymentStep:IStep<OrderContext>{privatereadonlyIPaymentService_payment;publicPaymentStep(IPaymentServicepayment)=>_payment=payment;publicasyncTask<StepResult>ExecuteAsync(OrderContextctx,CancellationTokenct=default){await_payment.ChargeAsync(ctx.Order,ct);ctx.Logs.Add("支付成功");returnStepResult.Continue;}}

Step 5 — 通知

publicclassNotifyStep:IStep<OrderContext>{privatereadonlyINotifier_notifier;publicNotifyStep(INotifiernotifier)=>_notifier=notifier;publicasyncTask<StepResult>ExecuteAsync(OrderContextctx,CancellationTokenct=default){await_notifier.SendAsync(ctx.Order,ct);ctx.Logs.Add("通知已发送");returnStepResult.Continue;}}

五、编排：从"硬编码"到"可配置"

方式 A：DI 容器自动装配（推荐）

// Program.csbuilder.Services.AddScoped<IStep<OrderContext>,ValidateOrderStep>();builder.Services.AddScoped<IStep<OrderContext>,CalculatePriceStep>();builder.Services.AddScoped<IStep<OrderContext>,DeductInventoryStep>();builder.Services.AddScoped<IStep<OrderContext>,PaymentStep>();builder.Services.AddScoped<IStep<OrderContext>,NotifyStep>();builder.Services.AddScoped<IPipeline<OrderContext>,Pipeline<OrderContext>>();

调用：

publicclassOrderAppService{privatereadonlyIPipeline<OrderContext>_pipeline;publicOrderAppService(IPipeline<OrderContext>pipeline)=>_pipeline=pipeline;publicasyncTaskPlaceOrderAsync(Orderorder){varctx=newOrderContext{Order=order};await_pipeline.ExecuteAsync(ctx);}}

⚠️IEnumerable<IStep<T>>的注入顺序取决于注册顺序。想严格控制顺序，看方式 B。

方式 B：显式命名 + 排序（更可控）

publicinterfaceIStep<TContext>{stringName{get;}intOrder{get;}Task<StepResult>ExecuteAsync(TContextcontext,CancellationTokenct=default);}

在Pipeline中排序：

_steps=steps.OrderBy(s=>s.Order).ToList();

方式 C：从配置/数据库动态生成管道（高级）

// pipeline.json{"steps":["ValidateOrderStep","CalculatePriceStep","DeductInventoryStep","PaymentStep","NotifyStep"]}

publicclassConfigurablePipeline<TContext>:IPipeline<TContext>{privatereadonlyIServiceProvider_sp;privatereadonlyList<Type>_stepTypes;publicConfigurablePipeline(IServiceProvidersp,IConfigurationcfg){_sp=sp;_stepTypes=cfg.GetSection("pipeline:steps").Get<List<string>>().Select(n=>Type.GetType($"MyApp.Steps.{n}")!).ToList();}publicasyncTaskExecuteAsync(TContextctx,CancellationTokenct=default){foreach(vartypein_stepTypes){varstep=(IStep<TContext>)_sp.GetRequiredService(type);if(awaitstep.ExecuteAsync(ctx,ct)==StepResult.Stop)break;}}}

这样换个 JSON 就换了流程，不用重新编译。

六、为什么满足你的 6 个特性？

单测示例（xUnit）

publicclassValidateOrderStepTests{[Fact]publicasyncTask 空订单_应终止流程(){varstep=newValidateOrderStep();varctx=newOrderContext{Order=newOrder()};// 无 itemsvarresult=awaitstep.ExecuteAsync(ctx);Assert.Equal(StepResult.Stop,result);Assert.False(ctx.IsValid);}[Fact]publicasyncTask 正常订单_应继续(){varstep=newValidateOrderStep();varctx=newOrderContext{Order=newOrder{Items={newOrderItem{Quantity=1}}}};varresult=awaitstep.ExecuteAsync(ctx);Assert.Equal(StepResult.Continue,result);Assert.True(ctx.IsValid);}}

完全不需要 mock 任何东西。

七、进阶：中间件风格（类似 ASP.NET Core）

如果 Step 需要"包裹"下一个 Step（做超时、日志、事务），可以用洋葱模型：

publicdelegateTaskStepDelegate<TContext>(TContextctx);publicinterfaceIMiddleware<TContext>{TaskInvokeAsync(TContextctx,StepDelegate<TContext>next);}publicclassPipelineBuilder<TContext>{privatereadonlyList<Func<StepDelegate<TContext>,StepDelegate<TContext>>>_components=new();publicPipelineBuilder<TContext>Use(Func<StepDelegate<TContext>,StepDelegate<TContext>>middleware){_components.Add(middleware);returnthis;}publicPipelineBuilder<TContext>UseStep<TS>(IServiceProvidersp)whereTS:IStep<TContext>{returnUse(next=>asyncctx=>{varstep=sp.GetRequiredService<TS>();varr=awaitstep.ExecuteAsync(ctx);if(r==StepResult.Continue)awaitnext(ctx);});}publicStepDelegate<TContext>Build(StepDelegate<TContext>terminal){varapp=terminal;for(inti=_components.Count-1;i>=0;i--)app=_components[i](app);returnapp;}}

用法（事务 + 日志 + 步骤）：

varpipeline=newPipelineBuilder<OrderContext>().Use(next=>asyncctx=>{usingvartx=_db.BeginTransaction();awaitnext(ctx);awaittx.CommitAsync();}).Use(next=>asyncctx=>{varsw=Stopwatch.StartNew();awaitnext(ctx);_log.Info($"耗时{sw.ElapsedMilliseconds}ms");}).UseStep<ValidateOrderStep>(sp).UseStep<CalculatePriceStep>(sp).UseStep<PaymentStep>(sp).Build(_=>Task.CompletedTask);awaitpipeline(ctx);

八、什么时候不该用？

• 步骤永远只有 2~3 个且不会变→ 直接用普通方法，别过度设计
• 步骤间强耦合、需要频繁互操作共享局部变量→ 管道反而更绕
• 团队不熟悉这种模式→ 先用注释清晰的顺序方法，等复杂了再重构

九、总结一句话

Pipeline =IEnumerable<IStep<TContext>>+ DI 排序 + 配置驱动。
每个 Step 一个类，管道负责顺序，配置负责编排——这就是可读、可维护、可测试、可复用、可编排、SOLID 的落地方式。

需要我针对某个具体业务（比如支付回调、审批流、数据 ETL）给出更贴合的管道设计吗？