C# 管道模式Pipeline Pattern详解一、核心思想你列出的六点本质上是责任链模式 组合模式 依赖注入的组合应用。核心就一句话把流程从一段又长又硬编码的方法拆解成一组可插拔的 Step由管道按顺序编排执行。传统写法 管道写法 DoEverything() Pipeline { ├── Step A stepA(); ├── Step B stepB(); ├── Step C stepC(); └── Step D }二、为什么需要它对比痛点❌ 传统写法的痛点publicclassOrderService{publicvoidPlaceOrder(Orderorder){// 1. 校验if(order.Items.Count0)thrownewException(空订单);// 2. 计算价格order.Totalorder.Items.Sum(ii.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;}publicListstringLogs{get;}new();}2. Step 接口 —— 每个步骤只做一件事publicinterfaceIStepTContext{TaskExecuteAsync(TContextcontext,CancellationTokenctdefault);}如果步骤需要决定是否继续往下走可以返回一个结果publicenumStepResult{Continue,Stop}publicinterfaceIStepTContext{TaskStepResultExecuteAsync(TContextcontext,CancellationTokenctdefault);}3. Pipeline —— 负责按顺序编排publicinterfaceIPipelineTContext{TaskExecuteAsync(TContextcontext,CancellationTokenctdefault);}publicclassPipelineTContext:IPipelineTContext{privatereadonlyIEnumerableIStepTContext_steps;publicPipeline(IEnumerableIStepTContextsteps){_stepssteps;}publicasyncTaskExecuteAsync(TContextcontext,CancellationTokenctdefault){foreach(varstepin_steps){varresultawaitstep.ExecuteAsync(context,ct);if(resultStepResult.Stop)break;}}}四、完整示例订单处理管道Step 1 — 校验publicclassValidateOrderStep:IStepOrderContext{publicTaskStepResultExecuteAsync(OrderContextctx,CancellationTokenctdefault){if(ctx.Order.Items.Count0){ctx.IsValidfalse;ctx.Logs.Add(订单为空终止流程);returnTask.FromResult(StepResult.Stop);}ctx.IsValidtrue;ctx.Logs.Add(校验通过);returnTask.FromResult(StepResult.Continue);}}Step 2 — 计算价格publicclassCalculatePriceStep:IStepOrderContext{publicTaskStepResultExecuteAsync(OrderContextctx,CancellationTokenctdefault){ctx.Order.Totalctx.Order.Items.Sum(ii.Price*i.Quantity)-ctx.Discount;ctx.Logs.Add($价格计算完成:{ctx.Order.Total});returnTask.FromResult(StepResult.Continue);}}Step 3 — 扣库存publicclassDeductInventoryStep:IStepOrderContext{privatereadonlyIInventoryService_inventory;publicDeductInventoryStep(IInventoryServiceinventory)_inventoryinventory;publicasyncTaskStepResultExecuteAsync(OrderContextctx,CancellationTokenctdefault){varokawait_inventory.DeductAsync(ctx.Order,ct);if(!ok){ctx.Logs.Add(库存不足终止流程);returnStepResult.Stop;}ctx.Logs.Add(库存扣减成功);returnStepResult.Continue;}}Step 4 — 支付publicclassPaymentStep:IStepOrderContext{privatereadonlyIPaymentService_payment;publicPaymentStep(IPaymentServicepayment)_paymentpayment;publicasyncTaskStepResultExecuteAsync(OrderContextctx,CancellationTokenctdefault){await_payment.ChargeAsync(ctx.Order,ct);ctx.Logs.Add(支付成功);returnStepResult.Continue;}}Step 5 — 通知publicclassNotifyStep:IStepOrderContext{privatereadonlyINotifier_notifier;publicNotifyStep(INotifiernotifier)_notifiernotifier;publicasyncTaskStepResultExecuteAsync(OrderContextctx,CancellationTokenctdefault){await_notifier.SendAsync(ctx.Order,ct);ctx.Logs.Add(通知已发送);returnStepResult.Continue;}}五、编排从硬编码到可配置方式 ADI 容器自动装配推荐// Program.csbuilder.Services.AddScopedIStepOrderContext,ValidateOrderStep();builder.Services.AddScopedIStepOrderContext,CalculatePriceStep();builder.Services.AddScopedIStepOrderContext,DeductInventoryStep();builder.Services.AddScopedIStepOrderContext,PaymentStep();builder.Services.AddScopedIStepOrderContext,NotifyStep();builder.Services.AddScopedIPipelineOrderContext,PipelineOrderContext();调用publicclassOrderAppService{privatereadonlyIPipelineOrderContext_pipeline;publicOrderAppService(IPipelineOrderContextpipeline)_pipelinepipeline;publicasyncTaskPlaceOrderAsync(Orderorder){varctxnewOrderContext{Orderorder};await_pipeline.ExecuteAsync(ctx);}}⚠️IEnumerableIStepT的注入顺序取决于注册顺序。想严格控制顺序看方式 B。方式 B显式命名 排序更可控publicinterfaceIStepTContext{stringName{get;}intOrder{get;}TaskStepResultExecuteAsync(TContextcontext,CancellationTokenctdefault);}在Pipeline中排序_stepssteps.OrderBy(ss.Order).ToList();方式 C从配置/数据库动态生成管道高级// pipeline.json{steps:[ValidateOrderStep,CalculatePriceStep,DeductInventoryStep,PaymentStep,NotifyStep]}publicclassConfigurablePipelineTContext:IPipelineTContext{privatereadonlyIServiceProvider_sp;privatereadonlyListType_stepTypes;publicConfigurablePipeline(IServiceProvidersp,IConfigurationcfg){_spsp;_stepTypescfg.GetSection(pipeline:steps).GetListstring().Select(nType.GetType($MyApp.Steps.{n})!).ToList();}publicasyncTaskExecuteAsync(TContextctx,CancellationTokenctdefault){foreach(vartypein_stepTypes){varstep(IStepTContext)_sp.GetRequiredService(type);if(awaitstep.ExecuteAsync(ctx,ct)StepResult.Stop)break;}}}这样换个 JSON 就换了流程不用重新编译。六、为什么满足你的 6 个特性特性如何满足可读性流程 IEnumerableIStep一眼看清单可维护新增步骤 新增一个类不动老代码可测试每个 Step 都是独立类输入 Context 输出 StepResult直接单测可复用NotifyStep可以同时被订单、退款、注册等管道复用可编排顺序由 DI 注册顺序 / 配置 / 数据库决定SOLID单一职责每 Step 一件事 开闭原则扩展加类修改改配置单测示例xUnitpublicclassValidateOrderStepTests{[Fact]publicasyncTask 空订单_应终止流程(){varstepnewValidateOrderStep();varctxnewOrderContext{OrdernewOrder()};// 无 itemsvarresultawaitstep.ExecuteAsync(ctx);Assert.Equal(StepResult.Stop,result);Assert.False(ctx.IsValid);}[Fact]publicasyncTask 正常订单_应继续(){varstepnewValidateOrderStep();varctxnewOrderContext{OrdernewOrder{Items{newOrderItem{Quantity1}}}};varresultawaitstep.ExecuteAsync(ctx);Assert.Equal(StepResult.Continue,result);Assert.True(ctx.IsValid);}}完全不需要 mock 任何东西。七、进阶中间件风格类似 ASP.NET Core如果 Step 需要包裹下一个 Step做超时、日志、事务可以用洋葱模型publicdelegateTaskStepDelegateTContext(TContextctx);publicinterfaceIMiddlewareTContext{TaskInvokeAsync(TContextctx,StepDelegateTContextnext);}publicclassPipelineBuilderTContext{privatereadonlyListFuncStepDelegateTContext,StepDelegateTContext_componentsnew();publicPipelineBuilderTContextUse(FuncStepDelegateTContext,StepDelegateTContextmiddleware){_components.Add(middleware);returnthis;}publicPipelineBuilderTContextUseStepTS(IServiceProvidersp)whereTS:IStepTContext{returnUse(nextasyncctx{varstepsp.GetRequiredServiceTS();varrawaitstep.ExecuteAsync(ctx);if(rStepResult.Continue)awaitnext(ctx);});}publicStepDelegateTContextBuild(StepDelegateTContextterminal){varappterminal;for(inti_components.Count-1;i0;i--)app_components[i](app);returnapp;}}用法事务 日志 步骤varpipelinenewPipelineBuilderOrderContext().Use(nextasyncctx{usingvartx_db.BeginTransaction();awaitnext(ctx);awaittx.CommitAsync();}).Use(nextasyncctx{varswStopwatch.StartNew();awaitnext(ctx);_log.Info($耗时{sw.ElapsedMilliseconds}ms);}).UseStepValidateOrderStep(sp).UseStepCalculatePriceStep(sp).UseStepPaymentStep(sp).Build(_Task.CompletedTask);awaitpipeline(ctx);八、什么时候不该用步骤永远只有 2~3 个且不会变→ 直接用普通方法别过度设计步骤间强耦合、需要频繁互操作共享局部变量→ 管道反而更绕团队不熟悉这种模式→ 先用注释清晰的顺序方法等复杂了再重构九、总结一句话Pipeline IEnumerableIStepTContext DI 排序 配置驱动。每个 Step 一个类管道负责顺序配置负责编排——这就是可读、可维护、可测试、可复用、可编排、SOLID 的落地方式。需要我针对某个具体业务比如支付回调、审批流、数据 ETL给出更贴合的管道设计吗
C# 管道模式Pipeline Pattern详解一、核心思想你列出的六点本质上是责任链模式 组合模式 依赖注入的组合应用。核心就一句话把流程从一段又长又硬编码的方法拆解成一组可插拔的 Step由管道按顺序编排执行。传统写法 管道写法 DoEverything() Pipeline { ├── Step A stepA(); ├── Step B stepB(); ├── Step C stepC(); └── Step D }二、为什么需要它对比痛点❌ 传统写法的痛点publicclassOrderService{publicvoidPlaceOrder(Orderorder){// 1. 校验if(order.Items.Count0)thrownewException(空订单);// 2. 计算价格order.Totalorder.Items.Sum(ii.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;}publicListstringLogs{get;}new();}2. Step 接口 —— 每个步骤只做一件事publicinterfaceIStepTContext{TaskExecuteAsync(TContextcontext,CancellationTokenctdefault);}如果步骤需要决定是否继续往下走可以返回一个结果publicenumStepResult{Continue,Stop}publicinterfaceIStepTContext{TaskStepResultExecuteAsync(TContextcontext,CancellationTokenctdefault);}3. Pipeline —— 负责按顺序编排publicinterfaceIPipelineTContext{TaskExecuteAsync(TContextcontext,CancellationTokenctdefault);}publicclassPipelineTContext:IPipelineTContext{privatereadonlyIEnumerableIStepTContext_steps;publicPipeline(IEnumerableIStepTContextsteps){_stepssteps;}publicasyncTaskExecuteAsync(TContextcontext,CancellationTokenctdefault){foreach(varstepin_steps){varresultawaitstep.ExecuteAsync(context,ct);if(resultStepResult.Stop)break;}}}四、完整示例订单处理管道Step 1 — 校验publicclassValidateOrderStep:IStepOrderContext{publicTaskStepResultExecuteAsync(OrderContextctx,CancellationTokenctdefault){if(ctx.Order.Items.Count0){ctx.IsValidfalse;ctx.Logs.Add(订单为空终止流程);returnTask.FromResult(StepResult.Stop);}ctx.IsValidtrue;ctx.Logs.Add(校验通过);returnTask.FromResult(StepResult.Continue);}}Step 2 — 计算价格publicclassCalculatePriceStep:IStepOrderContext{publicTaskStepResultExecuteAsync(OrderContextctx,CancellationTokenctdefault){ctx.Order.Totalctx.Order.Items.Sum(ii.Price*i.Quantity)-ctx.Discount;ctx.Logs.Add($价格计算完成:{ctx.Order.Total});returnTask.FromResult(StepResult.Continue);}}Step 3 — 扣库存publicclassDeductInventoryStep:IStepOrderContext{privatereadonlyIInventoryService_inventory;publicDeductInventoryStep(IInventoryServiceinventory)_inventoryinventory;publicasyncTaskStepResultExecuteAsync(OrderContextctx,CancellationTokenctdefault){varokawait_inventory.DeductAsync(ctx.Order,ct);if(!ok){ctx.Logs.Add(库存不足终止流程);returnStepResult.Stop;}ctx.Logs.Add(库存扣减成功);returnStepResult.Continue;}}Step 4 — 支付publicclassPaymentStep:IStepOrderContext{privatereadonlyIPaymentService_payment;publicPaymentStep(IPaymentServicepayment)_paymentpayment;publicasyncTaskStepResultExecuteAsync(OrderContextctx,CancellationTokenctdefault){await_payment.ChargeAsync(ctx.Order,ct);ctx.Logs.Add(支付成功);returnStepResult.Continue;}}Step 5 — 通知publicclassNotifyStep:IStepOrderContext{privatereadonlyINotifier_notifier;publicNotifyStep(INotifiernotifier)_notifiernotifier;publicasyncTaskStepResultExecuteAsync(OrderContextctx,CancellationTokenctdefault){await_notifier.SendAsync(ctx.Order,ct);ctx.Logs.Add(通知已发送);returnStepResult.Continue;}}五、编排从硬编码到可配置方式 ADI 容器自动装配推荐// Program.csbuilder.Services.AddScopedIStepOrderContext,ValidateOrderStep();builder.Services.AddScopedIStepOrderContext,CalculatePriceStep();builder.Services.AddScopedIStepOrderContext,DeductInventoryStep();builder.Services.AddScopedIStepOrderContext,PaymentStep();builder.Services.AddScopedIStepOrderContext,NotifyStep();builder.Services.AddScopedIPipelineOrderContext,PipelineOrderContext();调用publicclassOrderAppService{privatereadonlyIPipelineOrderContext_pipeline;publicOrderAppService(IPipelineOrderContextpipeline)_pipelinepipeline;publicasyncTaskPlaceOrderAsync(Orderorder){varctxnewOrderContext{Orderorder};await_pipeline.ExecuteAsync(ctx);}}⚠️IEnumerableIStepT的注入顺序取决于注册顺序。想严格控制顺序看方式 B。方式 B显式命名 排序更可控publicinterfaceIStepTContext{stringName{get;}intOrder{get;}TaskStepResultExecuteAsync(TContextcontext,CancellationTokenctdefault);}在Pipeline中排序_stepssteps.OrderBy(ss.Order).ToList();方式 C从配置/数据库动态生成管道高级// pipeline.json{steps:[ValidateOrderStep,CalculatePriceStep,DeductInventoryStep,PaymentStep,NotifyStep]}publicclassConfigurablePipelineTContext:IPipelineTContext{privatereadonlyIServiceProvider_sp;privatereadonlyListType_stepTypes;publicConfigurablePipeline(IServiceProvidersp,IConfigurationcfg){_spsp;_stepTypescfg.GetSection(pipeline:steps).GetListstring().Select(nType.GetType($MyApp.Steps.{n})!).ToList();}publicasyncTaskExecuteAsync(TContextctx,CancellationTokenctdefault){foreach(vartypein_stepTypes){varstep(IStepTContext)_sp.GetRequiredService(type);if(awaitstep.ExecuteAsync(ctx,ct)StepResult.Stop)break;}}}这样换个 JSON 就换了流程不用重新编译。六、为什么满足你的 6 个特性特性如何满足可读性流程 IEnumerableIStep一眼看清单可维护新增步骤 新增一个类不动老代码可测试每个 Step 都是独立类输入 Context 输出 StepResult直接单测可复用NotifyStep可以同时被订单、退款、注册等管道复用可编排顺序由 DI 注册顺序 / 配置 / 数据库决定SOLID单一职责每 Step 一件事 开闭原则扩展加类修改改配置单测示例xUnitpublicclassValidateOrderStepTests{[Fact]publicasyncTask 空订单_应终止流程(){varstepnewValidateOrderStep();varctxnewOrderContext{OrdernewOrder()};// 无 itemsvarresultawaitstep.ExecuteAsync(ctx);Assert.Equal(StepResult.Stop,result);Assert.False(ctx.IsValid);}[Fact]publicasyncTask 正常订单_应继续(){varstepnewValidateOrderStep();varctxnewOrderContext{OrdernewOrder{Items{newOrderItem{Quantity1}}}};varresultawaitstep.ExecuteAsync(ctx);Assert.Equal(StepResult.Continue,result);Assert.True(ctx.IsValid);}}完全不需要 mock 任何东西。七、进阶中间件风格类似 ASP.NET Core如果 Step 需要包裹下一个 Step做超时、日志、事务可以用洋葱模型publicdelegateTaskStepDelegateTContext(TContextctx);publicinterfaceIMiddlewareTContext{TaskInvokeAsync(TContextctx,StepDelegateTContextnext);}publicclassPipelineBuilderTContext{privatereadonlyListFuncStepDelegateTContext,StepDelegateTContext_componentsnew();publicPipelineBuilderTContextUse(FuncStepDelegateTContext,StepDelegateTContextmiddleware){_components.Add(middleware);returnthis;}publicPipelineBuilderTContextUseStepTS(IServiceProvidersp)whereTS:IStepTContext{returnUse(nextasyncctx{varstepsp.GetRequiredServiceTS();varrawaitstep.ExecuteAsync(ctx);if(rStepResult.Continue)awaitnext(ctx);});}publicStepDelegateTContextBuild(StepDelegateTContextterminal){varappterminal;for(inti_components.Count-1;i0;i--)app_components[i](app);returnapp;}}用法事务 日志 步骤varpipelinenewPipelineBuilderOrderContext().Use(nextasyncctx{usingvartx_db.BeginTransaction();awaitnext(ctx);awaittx.CommitAsync();}).Use(nextasyncctx{varswStopwatch.StartNew();awaitnext(ctx);_log.Info($耗时{sw.ElapsedMilliseconds}ms);}).UseStepValidateOrderStep(sp).UseStepCalculatePriceStep(sp).UseStepPaymentStep(sp).Build(_Task.CompletedTask);awaitpipeline(ctx);八、什么时候不该用步骤永远只有 2~3 个且不会变→ 直接用普通方法别过度设计步骤间强耦合、需要频繁互操作共享局部变量→ 管道反而更绕团队不熟悉这种模式→ 先用注释清晰的顺序方法等复杂了再重构九、总结一句话Pipeline IEnumerableIStepTContext DI 排序 配置驱动。每个 Step 一个类管道负责顺序配置负责编排——这就是可读、可维护、可测试、可复用、可编排、SOLID 的落地方式。需要我针对某个具体业务比如支付回调、审批流、数据 ETL给出更贴合的管道设计吗
)
平方复杂度无法被常规优化消除｜TaoToken 统一 Key 通道实测)
