Task Runner

ParallelProcessPractice.sln

@@ -47,6 +47,8 @@ Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "NathanDemo", "NathanDemo\Na
 EndProject
 Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "JolinDemo", "JolinDemo\JolinDemo.csproj", "{C58EA4C0-FFEC-43D6-97ED-EF755AE14B64}"
 EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "SusuDemo", "SusuDemo\SusuDemo.csproj", "{E96001F8-8734-4ECE-B8F9-80D58C95C4B6}"
+EndProject
 Global
 	GlobalSection(SolutionConfigurationPlatforms) = preSolution
 		Debug|Any CPU = Debug|Any CPU
@@ -117,6 +119,10 @@ Global
 		{C58EA4C0-FFEC-43D6-97ED-EF755AE14B64}.Debug|Any CPU.Build.0 = Debug|Any CPU
 		{C58EA4C0-FFEC-43D6-97ED-EF755AE14B64}.Release|Any CPU.ActiveCfg = Release|Any CPU
 		{C58EA4C0-FFEC-43D6-97ED-EF755AE14B64}.Release|Any CPU.Build.0 = Release|Any CPU
+		{E96001F8-8734-4ECE-B8F9-80D58C95C4B6}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{E96001F8-8734-4ECE-B8F9-80D58C95C4B6}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{E96001F8-8734-4ECE-B8F9-80D58C95C4B6}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{E96001F8-8734-4ECE-B8F9-80D58C95C4B6}.Release|Any CPU.Build.0 = Release|Any CPU
 	EndGlobalSection
 	GlobalSection(SolutionProperties) = preSolution
 		HideSolutionNode = FALSE
@@ -136,6 +142,7 @@ Global
 		{1B34613F-9384-4365-A54E-D9719D6E902B} = {B21D6D24-8EC2-497F-AE16-E0155FEE28CE}
 		{CD021A28-2C35-4FAA-BFDC-3E4543F009A0} = {B21D6D24-8EC2-497F-AE16-E0155FEE28CE}
 		{C58EA4C0-FFEC-43D6-97ED-EF755AE14B64} = {B21D6D24-8EC2-497F-AE16-E0155FEE28CE}
+		{E96001F8-8734-4ECE-B8F9-80D58C95C4B6} = {B21D6D24-8EC2-497F-AE16-E0155FEE28CE}
 	EndGlobalSection
 	GlobalSection(ExtensibilityGlobals) = postSolution
 		SolutionGuid = {15051360-3A56-4052-A944-97C62F90EEC6}

SusuDemo/Program.cs

@@ -0,0 +1,14 @@
+using System;
+using ParallelProcessPractice.Core;
+
+namespace SusuDemo
+{
+    class Program
+    {
+        static void Main(string[] args)
+        {
+            TaskRunnerBase run = new SusuTaskRunner();
+            run.ExecuteTasks(100);
+        }
+    }
+}

SusuDemo/SusuDemo.csproj

@@ -0,0 +1,12 @@
+<Project Sdk="Microsoft.NET.Sdk">
+
+  <PropertyGroup>
+    <OutputType>Exe</OutputType>
+    <TargetFramework>netcoreapp3.1</TargetFramework>
+  </PropertyGroup>
+
+  <ItemGroup>
+    <ProjectReference Include="..\ParallelProcessPractice.Core\ParallelProcessPractice.Core.csproj" />
+  </ItemGroup>
+
+</Project>

SusuDemo/SusuTaskRunner.cs

@@ -0,0 +1,130 @@
+using System;
+using System.Collections.Concurrent;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+using ParallelProcessPractice.Core;
+
+namespace SusuDemo
+{
+    public class SusuTaskRunner : TaskRunnerBase
+    {
+        private int blockingCollectionSize = 10;
+
+        public override void Run(IEnumerable<MyTask> tasks)
+        {
+            this.BlockingCollectionRun(tasks);
+        }
+
+        /// <summary>
+        /// 批次處理
+        /// </summary>
+        /// <param name="tasks">任務列表</param>
+        private void BatchRun(IEnumerable<MyTask> tasks)
+        {
+            foreach (var task in tasks)
+            {
+                task.DoStepN(1);
+            }
+
+            foreach (var task in tasks)
+            {
+                task.DoStepN(2);
+            }
+
+            foreach (var task in tasks)
+            {
+                task.DoStepN(3);
+            }
+        }
+
+        /// <summary>
+        /// 串流處理
+        /// </summary>
+        /// <param name="tasks">任務列表</param>
+        private void StreamRun(IEnumerable<MyTask> tasks)
+        {
+            foreach (var task in tasks)
+            {
+                task.DoStepN(1);
+                task.DoStepN(2);
+                task.DoStepN(3);
+            }
+        }
+
+        /// <summary>
+        /// 管線處理
+        /// </summary>
+        /// <param name="tasks">任務列表</param>
+        private void PipelinRun(IEnumerable<MyTask> tasks)
+        {
+            this.PipelinProcessStep(this.PipelinProcessStep(this.PipelinProcessStep(tasks, 1), 2), 3).ToList();
+        }
+
+        private IEnumerable<MyTask> PipelinProcessStep(IEnumerable<MyTask> tasks, int step)
+        {
+            foreach (var task in tasks)
+            {
+                task.DoStepN(step);
+                yield return task;
+            }
+        }
+
+        /// <summary>
+        /// 管線處理(Async)
+        /// </summary>
+        /// <param name="tasks">任務列表</param>
+        private void PipelineAsyncRun(IEnumerable<MyTask> tasks)
+        {
+            this.PipelinAsyncProcessStep(this.PipelinAsyncProcessStep(this.PipelinAsyncProcessStep(tasks, 1), 2), 3).ToList();
+        }
+
+        private IEnumerable<MyTask> PipelinAsyncProcessStep(IEnumerable<MyTask> tasks, int step)
+        {
+            Task<MyTask> taskResult = null;
+
+            foreach (var task in tasks)
+            {
+                if (taskResult != null) yield return taskResult.GetAwaiter().GetResult();
+                taskResult = Task.Run<MyTask>(() => 
+                {
+                    task.DoStepN(step);
+                    return task;
+                });
+            }
+
+            if (taskResult != null) yield return taskResult.GetAwaiter().GetResult();
+        }
+
+        /// <summary>
+        /// 管線處理(BlockingCollection)
+        /// </summary>
+        /// <param name="tasks">任務列表</param>
+        private void BlockingCollectionRun(IEnumerable<MyTask> tasks)
+        {
+            this.BlockingCollectionProcessStep(this.BlockingCollectionProcessStep(this.BlockingCollectionProcessStep(tasks, 1), 2), 3).ToList();
+        }
+
+        private IEnumerable<MyTask> BlockingCollectionProcessStep(IEnumerable<MyTask> tasks, int step)
+        {
+            BlockingCollection<MyTask> taskResultCollection = new BlockingCollection<MyTask>(blockingCollectionSize);
+
+            Task.Run(() => 
+            {
+                foreach (var task in tasks)
+                {
+                    task.DoStepN(step);
+                    taskResultCollection.Add(task);
+                }
+
+                taskResultCollection.CompleteAdding();
+            });
+
+            foreach (var task in taskResultCollection.GetConsumingEnumerable())
+            {
+                yield return task;
+            }
+        }
+    }
+}