使用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

复制代码

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


免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！