1. 首页
  2. .NET
  3. 正文

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

2020年4月30日 4739点热度 5人点赞 0条评论
内容目录

前言

前面学习了很多多线程和任务的基础知识，这里要来实践一下啦。通过本篇教程，你可以写出一个简单的工作流引擎。

本篇教程内容完成是基于任务的，只需要看过笔者的三篇关于异步的文章，掌握 C# 基础，即可轻松完成。

由于本篇文章编写的工作流程序，主要使用任务，有些逻辑过程会比较难理解，多测试一下就好。代码主要还是 C# 基础，为什么说简单？

  • 不包含 async 、await
  • 几乎不含包含多线程(有个读写锁)
  • 不包含表达式树
  • 几乎不含反射(有个小地方需要反射一下，但是非常简单)
  • 没有复杂的算法

因为是基于任务(Task)的，所以可以轻松设计组合流程，组成复杂的工作流。

由于只是讲述基础，所以不会包含很多种流程控制，这里只实现一些简单的。

先说明，别用到业务上。。。这个工作流非常简单，就几个功能，这个工作流是基于笔者的多线程系列文章的知识点。写这个东西是为了讲解任务操作，让读者更加深入理解任务。

代码地址：https://github.com/whuanle/CZGL.FLow

这两天忙着搬东西，今天没认真写文章，代码不明白的地方，可以到微信群找我。微信名称：痴者工良，dotnet 的群基本我都在。

节点

在开始前，我们来设计几种流程控制的东西。

将一个 步骤/流程/节点 称为 step。

Then

一个普通的节点，包含一个任务。

多个 Then 节点，可以组成一条连续的工作流。

file

Parallel

并行节点，可以设置多个并行节点放到 Parallel 中，以及在里面为任一个节点创建新的分支。

file

Schedule

定时节点，创建后会在一定时间后执行节点中的任务。

file

Delay

让当前任务阻塞一段时间。

file

试用一下

顺序节点

打开你的 VS ，创建项目，Nuget 引用 CZGL.DoFlow ，版本 1.0.2 。

创建一个类 MyFlow1，继承 IDoFlow。

    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 是构建工作流的一个接口。

我们来写一个工作流测试一下。

/// <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;
    }
}

Main 方法中：

        static void Main(string[] args)
        {
            FlowCore.RegisterWorkflow<myflow1>();
            // FlowCore.RegisterWorkflow(new MyFlow1());
            FlowCore.Start(1);
            Console.ReadKey();
        }</myflow1>

.StartWith() 方法开始一个工作流；

FlowCore.RegisterWorkflow<T>() 注册一个工作流；

FlowCore.Start();执行一个工作流；

并行任务

其代码如下：

    /// <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;
        }
    }

Main 方法中：

        static void Main(string[] args)
        {
            FlowCore.RegisterWorkflow<myflow2>();
            FlowCore.Start(2);
            Console.ReadKey();
        }</myflow2>

通过以上示例，可以大概了解本篇文章中我们要写的程序。

编写工作流

建立一个类库项目，名为 DoFlow。

建立 Extensions、Interfaces、Services 三个目录。

接口构建器

新建 IStepBuilder 接口文件到 Interfaces 目录，其内容如下：

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"></param>任意一个任务完成即可跳转到下一个step
        /// <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);
    }
}
</istepparallel>

新建 IStepParallel 文件到 Interfaces 目录。

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
    {

    }
}

工作流构建器

新建 IDoFlowBuilder 接口文件到 Interfaces 目录。

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; }
    }
}

新建 IDoFlow 接口文件到 Interfaces 目录。

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);
    }
}

依赖注入

新建 DependencyInjectionService 文件到 Services 目录。

用于实现依赖注入和解耦。

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;
        }
    }
}
</tiresult></tiresult></tservice></tservice>

添加一个 InitExtension 文件到 Extensions 目录。

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="">();
        }
    }
}
</istepparallelany></istepparallel></idoflowbuilder></istepbuilder>

实现工作流解析

以下文件均在 Services 目录建立。

新建 StepBuilder 文件，用于解析节点，构建任务。

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;
        }
    }
}
</istepparallel></istepparallelany></istepparallel>

新建 StepParallel 文件，里面有两个类，用于实现同步任务。

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;
        }
    }
}
</istepbuilder></task></task></istepbuilder></task></task>

新建 DoFlowBuilder 文件，用于构建工作流。

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;
        }
    }
}
</istepbuilder>

新建 FlowEngine 文件，用于执行工作流。

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();
        }
    }
}
</idoflowbuilder>

新建 FlowCore 文件，用于存储和索引工作流。使用读写锁解决并发字典问题。

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[id];
                }
                catch { return default; }
                finally
                {
                    readerWriterLockSlim.ExitWriteLock();
                }
            }
            catch { return default; }
            finally
            {
                readerWriterLockSlim.ExitUpgradeableReadLock();
            }

            engine.Start();
            return true;
        }
    }
}
</tdoflow></int></int>

就这样程序写完了。

忙去了。

本作品采用 知识共享署名-非商业性使用-相同方式共享 4.0 国际许可协议 进行许可
标签：
最后更新：2021年3月15日

痴者工良

高级程序员劝退师

文章评论

razz evil exclaim smile redface biggrin eek confused idea lol mad twisted rolleyes wink cool arrow neutral cry mrgreen drooling persevering

文章目录

COPYRIGHT © 2024 whuanle.cn. ALL RIGHTS RESERVED.

Theme Kratos Made By Seaton Jiang

粤ICP备18051778号

粤公网安备 44030902003257号