使用Pipeline抽象業務生命週期流程

　　上篇關於流程引擎的文章還是快兩年以前的《微服務業務生命週期流程管控引擎》，這中間各種低程式碼平臺層出不窮，雖然有些僅僅是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

如果你已經看到這裡，並且感覺還行的話可以在下方點個贊，或者也可以關注我的公總號（見二維條碼）