上篇關于流程引擎的文章還是快兩年以前的《微服務業務生命周期流程管控引擎》，這中間各種低代碼平臺層出不窮，雖然有些僅僅是OA+表單的再度包裝，但有些的確是在邏輯和操作單元層面進行了真正的高度抽象，形成產品底座，上層支持更大靈活程度的自定義應用搭建，當然這個不是本篇文章的主題，核心希望給各位朋友表達的還是對于邏輯流程抽象的關注，特別是現在的很多應用是在圍繞行業展開，如何在各種雜亂的功能點之中，做到雜而不亂，又如何通過核心的控制模塊提供標準化的接入，我通過OSS.Pipeline 這個引擎框架類別庫，和大家分享一下我的思路（如有不足，還望指出探討）,需要說明的是，此框架是類別庫級引擎，不依賴資料庫等存盤，代碼詳見Gitee或Github，
一. 我們抽象的是什么？
二. 邏輯推進和業務單元解耦思路
三. Pipeline 管道的設計實作
四. 使用Pipeline完成示例
一. 我們抽象的是什么？
首先，這個不是一個有著標準答案的問答題，只是用來開始這篇文章的發散思考，當然每個有著一定開發經驗的過來人也可能都有自己不同的見解，我也做開發多年，從早些年的單純CRUD（ctrl+c,ctrl+v 也干了不少），到后來參與復雜的業務邏輯，一直到自己全程負責打造產品，個人總結來看，常規業務產品的開發作業包含兩個方面：1. 單點業務的操作（依然是CRUD為主），處理資料的存盤和展示， 2. 點與點的連接（變數的部分），處理資料的流向，
單點的操作非常的簡單，就是諸如保存檔案，保存資料庫，第三方介面呼叫等等，很多專案的復雜，主要是集中在第二點，不同的業務有著不同的生命周期事件點組成，特別是再配合不同的公司管理流程加入，即便是相同的業務，系統資料的流向也各有不同，在我有限的經驗里，基本是在處理和抽象這一層面，（當然業務單元的粒度大小也是相當重要的，只是這個難度更容易解決）
二. 邏輯推進和業務單元解耦思路
如何解耦，這個每個人的方式方法多種多樣，我先以簡單的訂單支付成功并且需要發送短信和郵件為例，看下演進的程序：
在早期，訊息佇列還沒有大規模使用的時候，處理方式如圖
這個時候，功能完全是放在一塊的，開發簡單快速，但功能耦合且性能低下，隨著訊息佇列這些基礎解決方案的使用，我們進行第二版快速改造：
這個時候發送郵件和短信耗時的部分通過訊息佇列轉移至獨立的服務處理，暫時提升了性能可用問題，但因為郵件和短信因為引數內容不同，依然需要拼接訊息去操作兩個不同的佇列，耦合度依然存在，我們再更新第三版：
通過這一版，我們回到一個佇列，增添了訂單支付成功Hook服務，訂單支付更新的方法內部不需要關注任何其他邏輯細節，僅需添加成功事件佇列，邏輯進一步解耦，同時為后續的擴展提供了空間（在hook服務和具體的短信郵件之間依然可以通過佇列處理，這里不做進一步說明），
此時我們梳理一下當前的資料流：
雖然我們在第三版已經有了很大改善，但是我們可以看到，所有資訊流的流動，依然是依賴上一個節點的顯示呼叫，如果這里我們再添加錯誤重試，操作日志等需求，還是會或多或少的侵入業務代碼之中，所以我們能不能更進一步處理，比如訂單支付更新只關注更新，無需關注事件是通過訊息佇列還是異步執行緒傳遞給下游，成功HooK方法只需要關注訊息組裝分發，無需關注是否失敗重試，
按照設想，可以得出如下圖所示：
如果我們能通過上邊的圖示將個業務單元完全獨立出來，那么在每個業務單元之間可以隨時插入新的模塊而相互之間不受干擾，且能根據實際的情況進行例外的介入處理，這也是我設計OSS.Pipeline的初衷，
三. Pipeline 管道的設計實作
通過上邊的訂單支付演程序序，基本展示了我的基本思路，這里我們將示例再次簡化，方便繼續講解具體的實作
OSS.Pipeline 將所有的業務單元抽象為一個個節點，這些節點負責業務的具體執行，通過將這些Pipe組合形成業務的生命周期的流水線，即Pipeline，同時Pipeline本身也可作為一個獨立的Pipe參與更上一個層級的業務流程之中（即子流水線），通過將業務輸出和邏輯輸出的拆解，借助.Net 的泛型每一個管道都能定義獨立的業務輸入輸出，和邏輯輸入輸出引數（有時，邏輯輸入輸出和業務輸入輸出雖然相同，但代表的含義不同），因為OSS.Pipeline 是為了業務生命周期而設計，所以我參照了BPM中的組件命名方式，并擴展對應的組件基類供業務層選擇使用，具體的可用組件實作請參照gitee代碼介紹
下邊我會用上邊訂單的示例，來搭建一個Pipeline示例，
四. 使用Pipeline完成示例
1. 定義支付更新活動
public class OrderPayReq { public long OrderId { get; set; } public decimal PayMoney { get; set; } } /// <summary> /// 訂單支付管道 /// OrderPayReq - 業務輸入引數， bool - 業務輸出執行成功失敗， long - 邏輯輸出訂單Id /// </summary> internal class OrderPay : BaseActivity<OrderPayReq, bool, long> { protected override async Task<TrafficSignal<bool, long>> Executing(OrderPayReq para) { LogHelper.Info($"支付訂單（{para.OrderId}）金額：{para.PayMoney} 成功"); await Task.Delay(10); // 回傳執行成功，并告訴下級管道 訂單Id return new TrafficSignal<bool, long>(true, para.OrderId); } }
2. 定義支付成功后的Hook活動：
public class NotifyMsg { public string target { get; set; } public string content { get; set; } public bool is_sms { get; set; } // 假設不是短信就是郵件 } /// <summary> /// 支付Hook /// long-是上級管道傳入的訂單Id， bool - 業務輸出執行成功失敗， List<NotifyMsg> 需要發送的訊息串列 /// </summary> internal class PayHook : BaseActivity<long, bool, List<NotifyMsg>> { protected override async Task<TrafficSignal<bool, List<NotifyMsg>>> Executing(long para) { LogHelper.Info($"執行訂單（{para}）Hook"); await Task.Delay(10); var msgs = new List<NotifyMsg> { new NotifyMsg() {target = "管理員", content = $"訂單（{para}）支付成功，請注意發貨"}, new NotifyMsg() {target = "用戶", content = $"訂單（{para}）支付成功，已經入服務流程", is_sms = true} }; return new TrafficSignal<bool, List<NotifyMsg>>(true, msgs); } }
3. 定義發送活動
/// <summary> /// 發送服務 /// NotifyMsg - 上級管道傳遞的業務輸入引數， bool - 當前業務執行成功失敗 /// </summary> internal class Notify : BaseActivity<NotifyMsg, bool> { protected override async Task<TrafficSignal<bool>> Executing(NotifyMsg para) { LogHelper.Info($"發送{(para.is_sms?"短信":"郵件")}訊息 ：{para.target}:{para.content}"); await Task.Delay(10); return new TrafficSignal<bool>(true); } }
4，定義一個Pipeline，將上邊的管道串聯起來，同時定義一個Watcher，將管道執行程序中的事件記錄下來
internal class OrderPayPipeline { private static readonly OrderPay _pay = new OrderPay(); private static readonly PayHook _payHook = new PayHook(); private static readonly Notify _notify = new Notify(); static OrderPayPipeline() { _pay .AppendMsgFlow("order_pay_event") // 添加默認實作的異步訊息佇列中 .Append(_payHook) // 訊息佇列資料流向hook管道 .AppendMsgEnumerator() // Hook處理后有多條訊息，添加訊息列舉器 .Append(_notify); // 列舉后的單個訊息體流入發送節點 // 添加日志，通過創建流水線，給流水線添加Watcher，會自動給下邊的所有Pipe添加Watcher _pay.AsPipeline(_notify, new PipeLineOption() { Watcher = new FlowWatcher() },"OrderPayPipeline"); } // 作為對外暴露介面 public Task<bool> PayOrder(OrderPayReq req) { return _pay.Execute(req); } } public class FlowWatcher : IPipeLineWatcher { public Task PreCall(string pipeCode, PipeType pipeType, object input) { LogHelper.Info($"進入 {pipeCode} 管道", "PipePreCall", "PipelineWatcher"); return Task.CompletedTask; } public Task Executed(string pipeCode, PipeType pipeType, object input, WatchResult watchResult) { LogHelper.Info($"管道 {pipeCode} 執行結束，結束信號：{watchResult.signal}", "PipeExecuted", "PipelineWatcher"); return Task.CompletedTask; } public Task Blocked(string pipeCode, PipeType pipeType, object input, WatchResult watchResult) { LogHelper.Info($"管道 {pipeCode} 阻塞", "PipeBlocked", "PipelineWatcher"); return Task.CompletedTask; } }
5. 添加業務實際呼叫，這里使用單元測驗：
private static readonly OrderPayPipeline payLine = new OrderPayPipeline(); [TestMethod] public async Task TestOrder() { var payRes =await payLine.PayOrder(new OrderPayReq() {OrderId = 111, PayMoney = 1000.00m}); await Task.Delay(100); Assert.IsTrue(payRes); // 訂單支付更新結果 }
最后這里業務執行的日志如下：
2022-09-13 Code: Key: Detail:支付訂單（111）金額：1000.00 成功
2022-09-13 Code: Key: Detail:執行訂單（111）Hook
2022-09-13 Code: Key: Detail:發送郵件訊息 ：管理員:訂單（111）支付成功，請注意發貨
2022-09-13 Code: Key: Detail:發送短信訊息 ：用戶:訂單（111）支付成功，已經入服務流程
通過Watcher記錄操作日志如下：
2022-09-13 Code: Key:PipePreCall Detail:進入 SimpleMsgFlow`1 管道
2022-09-13 Code: Key:PipeExecuted Detail:管道 OrderPay 執行結束，結束信號：Green_Pass
2022-09-13 Code: Key:PipeExecuted Detail:管道 SimpleMsgFlow`1 執行結束，結束信號：Green_Pass
2022-09-13 Code: Key:PipePreCall Detail:進入 PayHook 管道
2022-09-13 Code: Key:PipeExecuted Detail:管道 PayHook 執行結束，結束信號：Green_Pass
2022-09-13 Code: Key:PipePreCall Detail:進入 MsgEnumerator`1 管道
2022-09-13 Code: Key:PipePreCall Detail:進入 Notify 管道
2022-09-13 Code: Key:PipePreCall Detail:進入 Notify 管道
2022-09-13 Code: Key:PipeExecuted Detail:管道 Notify 執行結束，結束信號：Green_Pass
2022-09-13 Code: Key:PipeExecuted Detail:管道 Notify 執行結束，結束信號：Green_Pass
2022-09-13 Code: Key:PipeExecuted Detail:管道 MsgEnumerator`1 執行結束，結束信號：Green_Pass
如果你已經看到這里，并且感徑訓行的話可以在下方點個贊，或者也可以關注我的公總號（見二維碼）
轉載請註明出處，本文鏈接：https://www.uj5u.com/qiye/508111.html
標籤：其他