C#多线程(16)：手把手教你撸一个工作流

[TOC] ## 前言 前面学习了很多多线程和任务的基础知识，这里要来实践一下啦。通过本篇教程，你可以写出一个简单的工作流引擎。 本篇教程内容完成是基于任务的，只需要看过笔者的三篇关于异步的文章，掌握 C# 基础，即可轻松完成。 * [C#多线程(13)：任务基础①](https://www.cnblogs.com/whuanle/p/12792639.html) * [C#多线程(14)：任务基础②](https://www.cnblogs.com/whuanle/p/12797269.html) * [C#多线程(15)：任务基础③](https://www.cnblogs.com/whuanle/p/12802943.html) 由于本篇文章编写的工作流程序，主要使用任务，有些逻辑过程会比较难理解，多测试一下就好。代码主要还是 C# 基础，为什么说简单？ * 不包含 async 、await * 几乎不含包含多线程(有个读写锁) * 不包含表达式树 * 几乎不含反射(有个小地方需要反射一下，但是非常简单) * 没有复杂的算法 因为是基于任务(Task)的，所以可以轻松设计组合流程，组成复杂的工作流。 由于只是讲述基础，所以不会包含很多种流程控制，这里只实现一些简单的。 先说明，别用到业务上。。。这个工作流非常简单，就几个功能，这个工作流是基于笔者的多线程系列文章的知识点。写这个东西是为了讲解任务操作，让读者更加深入理解任务。 代码地址：[https://github.com/whuanle/CZGL.FLow](https://github.com/whuanle/CZGL.FLow) 这两天忙着搬东西，今天没认真写文章，代码不明白的地方，可以到微信群找我。微信名称：痴者工良，dotnet 的群基本我都在。 ## 节点 在开始前，我们来设计几种流程控制的东西。 将一个 步骤/流程/节点 称为 step。 ### Then 一个普通的节点，包含一个任务。 多个 Then 节点，可以组成一条连续的工作流。  ### Parallel 并行节点，可以设置多个并行节点放到 Parallel 中，以及在里面为任一个节点创建新的分支。  ### Schedule 定时节点，创建后会在一定时间后执行节点中的任务。  ### Delay 让当前任务阻塞一段时间。  ## 试用一下 ### 顺序节点 打开你的 VS ，创建项目，Nuget 引用 `CZGL.DoFlow` ，版本 1.0.2 。 创建一个类 `MyFlow1`，继承 `IDoFlow`。 ```csharp public class MyFlow1 : IDoFlow { public int Id => 1; public string Name => "随便起个名字"; public int Version => 1; public IDoFlowBuilder Build(IDoFlowBuilder builder) { throw new NotImplementedException(); } } ``` 你可以创建多个工作流任务，每个工作流的 Id 必须唯一。Name 和 Version 随便填，因为这里笔者没有对这几个字段做逻辑。 `IDoFlowBuilder` 是构建工作流的一个接口。 我们来写一个工作流测试一下。 ```csharp /// /// 普通节点 Then 使用方法 /// public class MyFlow1 : IDoFlow { public int Id => 1; public string Name => "test"; public int Version => 1; public IDoFlowBuilder Build(IDoFlowBuilder builder) { builder.StartWith(() => { Console.WriteLine("工作流开始"); }).Then(() => { Console.WriteLine("下一个节点"); }).Then(() => { Console.WriteLine("最后一个节点"); }); return builder; } } ``` Main 方法中： ```csharp static void Main(string[] args) { FlowCore.RegisterWorkflow(); // FlowCore.RegisterWorkflow(new MyFlow1()); FlowCore.Start(1); Console.ReadKey(); } ``` `.StartWith()` 方法开始一个工作流； `FlowCore.RegisterWorkflow()` 注册一个工作流； `FlowCore.Start();`执行一个工作流； ### 并行任务 其代码如下： ```csharp /// /// 并行节点 Parallel 使用方法 /// public class MyFlow2 : IDoFlow { public int Id => 2; public string Name => "test"; public int Version => 1; public IDoFlowBuilder Build(IDoFlowBuilder builder) { builder.StartWith() .Parallel(steps => { // 每个并行任务也可以设计后面继续执行其它任务 steps.Do(() => { Console.WriteLine("并行1"); }).Do(() => { Console.WriteLine("并行2"); }); steps.Do(() => { Console.WriteLine("并行3"); }); // 并行任务设计完成后，必须调用此方法 // 此方法必须放在所有并行任务 .Do() 的最后 steps.EndParallel(); // 如果 .Do() 在 EndParallel() 后，那么不会等待此任务 steps.Do(() => { Console.WriteLine("并行异步"); }); // 开启新的分支 steps.StartWith() .Then(() => { Console.WriteLine("新的分支" + Task.CurrentId); }).Then(() => { Console.WriteLine("分支2.0" + Task.CurrentId); }); }, false) .Then(() => { Console.WriteLine("11111111111111111 "); }); return builder; } } ``` Main 方法中： ```csharp static void Main(string[] args) { FlowCore.RegisterWorkflow(); FlowCore.Start(2); Console.ReadKey(); } ``` 通过以上示例，可以大概了解本篇文章中我们要写的程序。 ## 编写工作流 建立一个类库项目，名为 `DoFlow`。 建立 `Extensions`、`Interfaces`、`Services` 三个目录。 ### 接口构建器 新建 `IStepBuilder` 接口文件到 `Interfaces` 目录，其内容如下： ```csharp using System; namespace DoFlow.Interfaces { public interface IStepBuilder { /// /// 普通节点 /// /// /// IStepBuilder Then(Action action); /// /// 多个节点 /// 默认下，需要等待所有的任务完成，这个step才算完成 /// /// /// 任意一个任务完成即可跳转到下一个step /// IStepBuilder Parallel(Action action, bool anyWait = false); /// /// 节点将在某个时间间隔后执行 /// 异步，不会阻塞当前工作流的运行，计划任务将在一段时间后触发 /// /// IStepBuilder Schedule(Action action, TimeSpan time); /// /// 阻塞一段时间 /// /// /// IStepBuilder Delay(TimeSpan time); } } ``` 新建 `IStepParallel` 文件到 `Interfaces` 目录。 ```csharp using System; namespace DoFlow.Interfaces { /// /// 并行任务 /// 默认情况下，只有这个节点的所有并行任务都完成后，这个节点才算完成 /// public interface IStepParallel { /// /// 一个并行任务 /// /// /// IStepParallel Do(Action action); /// /// 开始一个分支 /// /// /// IStepBuilder StartWith(Action action = null); /// /// 必须使用此方法结束一个并行任务 /// void EndParallel(); } /// /// 并行任务 /// 任意一个任务完成后，就可以跳转到下一个 step /// public interface IStepParallelAny : IStepParallel { } } ``` ### 工作流构建器 新建 `IDoFlowBuilder` 接口文件到 `Interfaces` 目录。 ```csharp using System; using System.Threading.Tasks; namespace DoFlow.Interfaces { /// /// 构建工作流任务 /// public interface IDoFlowBuilder { /// /// 开始一个 step /// IStepBuilder StartWith(Action action = null); void EndWith(Action action); Task ThatTask { get; } } } ``` 新建 `IDoFlow` 接口文件到 `Interfaces` 目录。 ```csharp namespace DoFlow.Interfaces { /// /// 工作流 /// 无参数传递 /// public interface IDoFlow { /// /// 全局唯一标识 /// int Id { get; } /// /// 标识此工作流的名称 /// string Name { get; } /// /// 标识此工作流的版本 /// int Version { get; } IDoFlowBuilder Build(IDoFlowBuilder builder); } } ``` ### 依赖注入 新建 `DependencyInjectionService` 文件到 `Services` 目录。 用于实现依赖注入和解耦。 ```csharp using DoFlow.Extensions; using Microsoft.Extensions.DependencyInjection; using System; namespace DoFlow.Services { /// /// 依赖注入服务 /// public static class DependencyInjectionService { private static IServiceCollection _servicesList; private static IServiceProvider _services; static DependencyInjectionService() { IServiceCollection services = new ServiceCollection(); _servicesList = services; // 注入引擎需要的服务 InitExtension.StartInitExtension(); var serviceProvider = services.BuildServiceProvider(); _services = serviceProvider; } /// /// 添加一个注入到容器服务 /// /// /// public static void AddService() where TService : class where TImplementation : class, TService { _servicesList.AddTransient(); } /// /// 获取需要的服务 /// /// /// public static TIResult GetService() { TIResult Tservice = _services.GetService(); return Tservice; } } } ``` 添加一个 `InitExtension` 文件到 `Extensions` 目录。 ```csharp using DoFlow.Interfaces; using DoFlow.Services; namespace DoFlow.Extensions { public static class InitExtension { private static bool IsInit = false; public static void StartInitExtension() { if (IsInit) return; IsInit = true; DependencyInjectionService.AddService(); DependencyInjectionService.AddService(); DependencyInjectionService.AddService(); DependencyInjectionService.AddService(); } } } ``` ### 实现工作流解析 以下文件均在 `Services` 目录建立。 新建 `StepBuilder` 文件，用于解析节点，构建任务。 ```csharp using DoFlow.Interfaces; using System; using System.Threading.Tasks; namespace DoFlow.Services { /// /// 节点工作引擎 /// public class StepBuilder : IStepBuilder { private Task _task; /// /// 延迟执行 /// /// /// public IStepBuilder Delay(TimeSpan time) { Task.Delay(time).Wait(); return this; } /// /// 并行 step /// /// /// public IStepBuilder Parallel(Action action, bool anyAwait = false) { IStepParallel parallel = anyAwait ? DependencyInjectionService.GetService() : DependencyInjectionService.GetService(); Task task = new Task(() => { action.Invoke(parallel); }); _task.ConfigureAwait(false).GetAwaiter().OnCompleted(() => { task.Start(); }); _task = task; return this; } /// /// 计划任务 /// /// /// /// public IStepBuilder Schedule(Action action, TimeSpan time) { Task.Factory.StartNew(() => { Task.Delay(time).Wait(); action.Invoke(); }); return this; } /// /// 普通 step /// /// /// public IStepBuilder Then(Action action) { Task task = new Task(action); _task.ConfigureAwait(false).GetAwaiter().OnCompleted(() => { task.Start(); task.Wait(); }); _task = task; return this; } public void SetTask(Task task) { _task = task; } } } ``` 新建 `StepParallel` 文件，里面有两个类，用于实现同步任务。 ```csharp using DoFlow.Interfaces; using System; using System.Collections.Generic; using System.Threading.Tasks; namespace DoFlow.Services { /// /// 第一层所有任务结束后才能跳转下一个 step /// public class StepParallelWhenAll : IStepParallel { private Task _task; private readonly List _tasks = new List(); public StepParallelWhenAll() { _task = new Task(() => { },TaskCreationOptions.AttachedToParent); } public IStepParallel Do(Action action) { _tasks.Add(Task.Run(action)); return this; } public void EndParallel() { _task.ConfigureAwait(false).GetAwaiter().OnCompleted(() => { Task.WhenAll(_tasks).Wait(); }); } public IStepBuilder StartWith(Action action = null) { Task task = action is null ? new Task(() => { }) : new Task(action); var _stepBuilder = DependencyInjectionService.GetService(); _task.ConfigureAwait(false).GetAwaiter().OnCompleted(() => { task.Start(); }); return _stepBuilder; } } /// /// 完成任意一个任务即可跳转到下一个 step /// public class StepParallelWhenAny : IStepParallelAny { private Task _task; private readonly List _tasks = new List(); public StepParallelWhenAny() { _task = Task.Run(() => { }); } public IStepParallel Do(Action action) { _tasks.Add(Task.Run(action)); return this; } public void EndParallel() { _task.ConfigureAwait(false).GetAwaiter().OnCompleted(() => { Task.WhenAny(_tasks).Wait(); }); } public IStepBuilder StartWith(Action action = null) { Task task = action is null ? new Task(() => { }) : new Task(action); var _stepBuilder = DependencyInjectionService.GetService(); _task.ConfigureAwait(false).GetAwaiter().OnCompleted(() => { task.Start(); }); return _stepBuilder; } } } ``` 新建 `DoFlowBuilder` 文件，用于构建工作流。 ```csharp using DoFlow.Interfaces; using System; using System.Threading.Tasks; namespace DoFlow.Services { public class DoFlowBuilder : IDoFlowBuilder { private Task _task; public Task ThatTask => _task; public void EndWith(Action action) { _task.Start(); } public IStepBuilder StartWith(Action action = null) { if (action is null) _task = new Task(() => { }); else _task = new Task(action); IStepBuilder _stepBuilder = DependencyInjectionService.GetService(); ((StepBuilder)_stepBuilder).SetTask(_task); return _stepBuilder; } } } ``` 新建 `FlowEngine` 文件，用于执行工作流。 ```csharp using DoFlow.Interfaces; namespace DoFlow.Services { /// /// 工作流引擎 /// public class FlowEngine { private readonly IDoFlow _flow; public FlowEngine(IDoFlow flow) { _flow = flow; } /// /// 开始一个工作流 /// public void Start() { IDoFlowBuilder builder = DependencyInjectionService.GetService(); _flow.Build(builder).ThatTask.Start(); } } } ``` 新建 `FlowCore` 文件，用于存储和索引工作流。使用读写锁解决并发字典问题。 ```csharp using DoFlow.Interfaces; using System; using System.Collections.Generic; using System.Threading; namespace DoFlow.Services { public static class FlowCore { private static Dictionary flowEngines = new Dictionary(); // 读写锁 private static ReaderWriterLockSlim readerWriterLockSlim = new ReaderWriterLockSlim(); /// /// 注册工作流 /// /// public static bool RegisterWorkflow(IDoFlow flow) { try { readerWriterLockSlim.EnterReadLock(); if (flowEngines.ContainsKey(flow.Id)) return false; flowEngines.Add(flow.Id, new FlowEngine(flow)); return true; } finally { readerWriterLockSlim.ExitReadLock(); } } /// /// 注册工作流 /// /// public static bool RegisterWorkflow() { Type type = typeof(TDoFlow); IDoFlow flow = (IDoFlow)Activator.CreateInstance(type); try { readerWriterLockSlim.EnterReadLock(); if (flowEngines.ContainsKey(flow.Id)) return false; flowEngines.Add(flow.Id, new FlowEngine(flow)); return true; } finally { readerWriterLockSlim.ExitReadLock(); } } /// /// 要启动的工作流 /// /// public static bool Start(int id) { FlowEngine engine; // 读写锁 try { readerWriterLockSlim.EnterUpgradeableReadLock(); if (!flowEngines.ContainsKey(id)) return default; try { readerWriterLockSlim.EnterWriteLock(); engine = flowEngines[id]; } catch { return default; } finally { readerWriterLockSlim.ExitWriteLock(); } } catch { return default; } finally { readerWriterLockSlim.ExitUpgradeableReadLock(); } engine.Start(); return true; } } } ``` 就这样程序写完了。 忙去了。