C# 管道模式Pipeline Pattern詳解一、核心思想你列出的六點本質上是責任鏈模式 組合模式 依賴注入的組合應用。核心就一句話把流程從一段又長又硬編碼的方法拆解成一組可插拔的 Step由管道按順序編排執(zhí)行。傳統(tǒng)寫法 管道寫法 DoEverything() Pipeline { ├── Step A stepA(); ├── Step B stepB(); ├── Step C stepC(); └── Step D }二、為什么需要它對比痛點? 傳統(tǒng)寫法的痛點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. 發(fā)通知_email.Send(order);// 6. 記錄日志_log.Info(下單成功);// 7. 后續(xù)還要加風控、優(yōu)惠券、積分...}}問題一個方法越來越長可讀性差加/刪步驟要改這個方法違反開閉原則沒法單獨測庫存扣減這一步難測試其他業(yè)務想復用發(fā)通知邏輯只能復制粘貼難復用三、核心抽象1. 上下文Context—— 在 Step 之間傳遞數(shù)據(jù)publicclassOrderContext{publicOrderOrder{get;set;}new();publicboolIsValid{get;set;}publicdecimalDiscount{get;set;}publicListstringLogs{get;}new();}2. Step 接口 —— 每個步驟只做一件事publicinterfaceIStepTContext{TaskExecuteAsync(TContextcontext,CancellationTokenctdefault);}如果步驟需要決定是否繼續(xù)往下走可以返回一個結果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(通知已發(fā)送);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從配置/數(shù)據(jù)庫動態(tài)生成管道高級// 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 注冊順序 / 配置 / 數(shù)據(jù)庫決定SOLID單一職責每 Step 一件事 開閉原則擴展加類修改改配置單測示例xUnitpublicclassValidateOrderStepTests{[Fact]publicasyncTask 空訂單_應終止流程(){varstepnewValidateOrderStep();varctxnewOrderContext{OrdernewOrder()};// 無 itemsvarresultawaitstep.ExecuteAsync(ctx);Assert.Equal(StepResult.Stop,result);Assert.False(ctx.IsValid);}[Fact]publicasyncTask 正常訂單_應繼續(xù)(){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 的落地方式。需要我針對某個具體業(yè)務比如支付回調、審批流、數(shù)據(jù) ETL給出更貼合的管道設計嗎 C# 管道模式（Pipeline Pattern）詳解
C# 管道模式Pipeline Pattern詳解一、核心思想你列出的六點本質上是責任鏈模式 組合模式 依賴注入的組合應用。核心就一句話把流程從一段又長又硬編碼的方法拆解成一組可插拔的 Step由管道按順序編排執(zhí)行。傳統(tǒng)寫法 管道寫法 DoEverything() Pipeline { ├── Step A stepA(); ├── Step B stepB(); ├── Step C stepC(); └── Step D }二、為什么需要它對比痛點? 傳統(tǒng)寫法的痛點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. 發(fā)通知_email.Send(order);// 6. 記錄日志_log.Info(下單成功);// 7. 后續(xù)還要加風控、優(yōu)惠券、積分...}}問題一個方法越來越長可讀性差加/刪步驟要改這個方法違反開閉原則沒法單獨測庫存扣減這一步難測試其他業(yè)務想復用發(fā)通知邏輯只能復制粘貼難復用三、核心抽象1. 上下文Context—— 在 Step 之間傳遞數(shù)據(jù)publicclassOrderContext{publicOrderOrder{get;set;}new();publicboolIsValid{get;set;}publicdecimalDiscount{get;set;}publicListstringLogs{get;}new();}2. Step 接口 —— 每個步驟只做一件事publicinterfaceIStepTContext{TaskExecuteAsync(TContextcontext,CancellationTokenctdefault);}如果步驟需要決定是否繼續(xù)往下走可以返回一個結果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(通知已發(fā)送);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從配置/數(shù)據(jù)庫動態(tài)生成管道高級// 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 注冊順序 / 配置 / 數(shù)據(jù)庫決定SOLID單一職責每 Step 一件事 開閉原則擴展加類修改改配置單測示例xUnitpublicclassValidateOrderStepTests{[Fact]publicasyncTask 空訂單_應終止流程(){varstepnewValidateOrderStep();varctxnewOrderContext{OrdernewOrder()};// 無 itemsvarresultawaitstep.ExecuteAsync(ctx);Assert.Equal(StepResult.Stop,result);Assert.False(ctx.IsValid);}[Fact]publicasyncTask 正常訂單_應繼續(xù)(){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 的落地方式。需要我針對某個具體業(yè)務比如支付回調、審批流、數(shù)據(jù) ETL給出更貼合的管道設計嗎