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