利用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. 定义支付更新活动

1.     public class OrderPayReq
2.     {
3.         public long OrderId { get; set; }
4.         public decimal PayMoney { get; set; }
5.     }
7.     /// <summary>
8.     ///  订单支付管道
9.     ///    OrderPayReq - 业务输入参数，  bool - 业务输出执行成功失败，   long - 逻辑输出订单Id
10.     /// </summary>
11.     internal class OrderPay : BaseActivity<OrderPayReq, bool, long>
12.     {
13.         protected override async Task<TrafficSignal<bool, long>> Executing(OrderPayReq para)
14.         {
15.             LogHelper.Info($"支付订单（{para.OrderId}）金额：{para.PayMoney} 成功");
17.             await Task.Delay(10);
19.             // 返回执行成功，并告诉下级管道 订单Id
20.             return new TrafficSignal<bool, long>(true, para.OrderId);
21.         }
22.     }<br>

复制代码

2. 定义支付乐成后的Hook活动：

1.     public class NotifyMsg
2.     {
3.         public string target { get; set; }
4.         public string content { get; set; }
5.         public bool is_sms { get; set; } // 假设不是短信就是邮件
6.     }
8.     /// <summary>
9.     ///  支付Hook
10.     ///     long-是上级管道传入的订单Id， bool - 业务输出执行成功失败，  List<NotifyMsg> 需要发送的消息列表
11.     /// </summary>
12.     internal class PayHook : BaseActivity<long, bool, List<NotifyMsg>>
13.     {
14.         protected override async Task<TrafficSignal<bool, List<NotifyMsg>>> Executing(long para)
15.         {
16.             LogHelper.Info($"执行订单（{para}）Hook");
17.             await Task.Delay(10);
19.             var msgs = new List<NotifyMsg>
20.             {
21.                 new NotifyMsg() {target = "管理员", content = $"订单（{para}）支付成功，请注意发货"},
22.                 new NotifyMsg() {target = "用户", content  = $"订单（{para}）支付成功，已经入服务流程", is_sms = true}
23.             };
25.             return new TrafficSignal<bool, List<NotifyMsg>>(true, msgs);
26.         }
27.     }

复制代码

3. 定义发送活动

1.     /// <summary>
2.     ///  发送服务
3.     ///     NotifyMsg - 上级管道传递的业务输入参数，   bool - 当前业务执行成功失败
4.     /// </summary>
5.     internal class Notify : BaseActivity<NotifyMsg, bool>
6.     {
7.         protected override async Task<TrafficSignal<bool>> Executing(NotifyMsg para)
8.         {
9.             LogHelper.Info($"发送{(para.is_sms?"短信":"邮件")}消息 ：{para.target}:{para.content}");
11.             await Task.Delay(10);
13.             return new TrafficSignal<bool>(true);
14.         }
15.     }

复制代码

4，定义一个Pipeline，将上边的管道串联起来，同时定义一个Watcher，将管道执行过程中的事件记录下来

1.     internal class OrderPayPipeline
2.     {
3.         private static readonly OrderPay _pay     = new OrderPay();
4.         private static readonly PayHook  _payHook = new PayHook();
5.         private static readonly Notify   _notify  = new Notify();
7.         static OrderPayPipeline()
8.         {
9.             _pay
10.                 .AppendMsgFlow("order_pay_event") // 添加默认实现的异步消息队列中
11.                 .Append(_payHook)                 // 消息队列数据流向hook管道
12.                 .AppendMsgEnumerator()            // Hook处理后有多条消息，添加消息枚举器
13.                 .Append(_notify);                 //  枚举后的单个消息体流入发送节点
15.             // 添加日志，通过创建流水线，给流水线添加Watcher，会自动给下边的所有Pipe添加Watcher
16.             _pay.AsPipeline(_notify, new PipeLineOption() { Watcher = new FlowWatcher() },"OrderPayPipeline");
17.         }
19.         // 作为对外暴露接口
20.         public Task<bool> PayOrder(OrderPayReq req)
21.         {
22.             return _pay.Execute(req);
23.         }
24.     }
26.     public class FlowWatcher : IPipeLineWatcher
27.     {
28.         public Task PreCall(string pipeCode, PipeType pipeType, object input)
29.         {
30.             LogHelper.Info($"进入 {pipeCode} 管道", "PipePreCall", "PipelineWatcher");
31.             return Task.CompletedTask;
32.         }
34.         public Task Executed(string pipeCode, PipeType pipeType, object input, WatchResult watchResult)
35.         {
36.             LogHelper.Info($"管道 {pipeCode} 执行结束，结束信号：{watchResult.signal}", "PipeExecuted", "PipelineWatcher");
37.             return Task.CompletedTask;
38.         }
40.         public Task Blocked(string pipeCode, PipeType pipeType, object input, WatchResult watchResult)
41.         {
42.             LogHelper.Info($"管道 {pipeCode} 阻塞", "PipeBlocked", "PipelineWatcher");
43.             return Task.CompletedTask;
44.         }
45.     }

复制代码

5. 添加业务实际调用，这里利用单元测试：

1.         private static readonly OrderPayPipeline payLine = new OrderPayPipeline();
3.         [TestMethod]
4.         public async Task TestOrder()
5.         {
6.             var payRes =await payLine.PayOrder(new OrderPayReq() {OrderId = 111, PayMoney = 1000.00m});
7.             await Task.Delay(100);
9.             Assert.IsTrue(payRes); // 订单支付更新结果
10.         }

复制代码

 末了这里业务执行的日志如下：

1. 2022-09-13    Code:    Key:   Detail:支付订单（111）金额：1000.00 成功
2. 2022-09-13    Code:    Key:   Detail:执行订单（111）Hook
3. 2022-09-13    Code:    Key:   Detail:发送邮件消息 ：管理员:订单（111）支付成功，请注意发货
4. 2022-09-13     Code:    Key:   Detail:发送短信消息 ：用户:订单（111）支付成功，已经入服务流程

复制代码

通过Watcher记录操作日志如下：

1. 2022-09-13   Code:    Key:PipePreCall   Detail:进入 SimpleMsgFlow`1 管道
2. 2022-09-13   Code:    Key:PipeExecuted   Detail:管道 OrderPay 执行结束，结束信号：Green_Pass
3. 2022-09-13   Code:    Key:PipeExecuted   Detail:管道 SimpleMsgFlow`1 执行结束，结束信号：Green_Pass
4. 2022-09-13   Code:    Key:PipePreCall   Detail:进入 PayHook 管道
5. 2022-09-13   Code:    Key:PipeExecuted   Detail:管道 PayHook 执行结束，结束信号：Green_Pass
6. 2022-09-13   Code:    Key:PipePreCall   Detail:进入 MsgEnumerator`1 管道
7. 2022-09-13   Code:    Key:PipePreCall   Detail:进入 Notify 管道
8. 2022-09-13   Code:    Key:PipePreCall   Detail:进入 Notify 管道
9. 2022-09-13   Code:    Key:PipeExecuted   Detail:管道 Notify 执行结束，结束信号：Green_Pass
10. 2022-09-13   Code:    Key:PipeExecuted   Detail:管道 Notify 执行结束，结束信号：Green_Pass
11. 2022-09-13   Code:    Key:PipeExecuted   Detail:管道 MsgEnumerator`1 执行结束，结束信号：Green_Pass

复制代码

 
假如你已经看到这里，并且感觉还行的话可以在下方点个赞，或者也可以关注我的公总号（见二维码）
 

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。