C# Task 多任务限制Task并发数量

LimitedTaskScheduler：

using System;
 System.Collections.Concurrent;
 System.Collections.Generic;
 System.Linq;
 System.Runtime.InteropServices;
 System.Text;
 System.Threading;
 System.Threading.Tasks;

namespace Utils
{
    public class LimitedTaskScheduler : TaskScheduler,Idisposable
    {
        #region 外部方法
        [DllImport("kernel32.dll",EntryPoint = SetProcessWorkingSetSize")]
        static extern int SetProcessWorkingSetSize(IntPtr process,int minSize,1)">int maxSize);
        #endregion

        #region 变量属性事件
        private BlockingCollection<Task> _tasks = new BlockingCollection<Task>();
        List<Thread> _threadList = new List<Thread>();
        private int _threadCount = 0;
        int _timeOut = Timeout.Infinite;
        private Task _tempTask;
        #region 构造函数
        public LimitedTaskScheduler(int threadCount = 10)
        {
            CreateThreads(threadCount);
        }
        #region override GetScheduledTasks
        protected override IEnumerable<Task> GetScheduledTasks()
        {
            return _tasks;
        }
        #region override TryExecuteTaskInline
        override bool TryExecuteTaskInline(Task task,1)">bool taskwasprevIoUslyQueued)
        {
            return false;
        }
        #region override QueueTask
        void QueueTask(Task task)
        {
            _tasks.Add(task);
        }
        #region 资源释放
        /// <summary>
        /// 资源释放
         如果尚有任务在执行，则会在调用此方法的线程上引发System.Threading.ThreadAbortException，请使用Task.WaitAll等待任务执行完毕后，再调用该方法
        </summary>
         dispose()
        {
            _timeOut = 100;

            foreach (Thread item in _threadList)
            {
                item.Abort();
            }
            _threadList.Clear();

            GC.Collect();
            GC.WaitForPendingFinalizers();
            if (Environment.Osversion.Platform == PlatformID.Win32NT)
            {
                SetProcessWorkingSetSize(System.Diagnostics.Process.GetCurrentProcess().Handle,-1,-1);
            }
        }
        #region 创建线程池
         创建线程池
        void CreateThreads(int? threadCount = null)
        {
            if (threadCount != null) _threadCount = threadCount.Value;
            _timeOut = Timeout.Infinite;

            for (int i = 0; i < _threadCount; i++)
            {
                Thread thread = new Thread(new ThreadStart(() =>
                {
                    Task task;
                    while (_tasks.TryTake(out task,_timeOut))
                    {
                        TryExecuteTask(task);
                    }
                }));
                thread.IsBackground = true;
                thread.Start();
                _threadList.Add(thread);
            }
        }
        #region 全部取消
         全部取消
         CancelAll()
        {
             _tempTask)) { }
        }
        #endregion

    }
}

View Code

ThreadHelper(Run方法没有使用LimitedTaskScheduler，Run2方法使用了LimitedTaskScheduler)：

 System.Windows.Threading;
 Utils
{
    <summary>
     线程帮助类(处理单线程任务)
    </summary>
     ThreadHelper
    {
        static LimitedTaskScheduler _defaultScheduler = new LimitedTaskScheduler();

         执行 
         例：ThreadHelper.Run(() => { },(ex) => { });
        </summary>
        <param name="doWork">在线程中执行</param>
        <param name="errorAction">错误处理</param>
        static System.Threading.Tasks.Task Run2(Action doWork,LimitedTaskScheduler scheduler = null,Action<Exception> errorAction = if (scheduler == null) scheduler = _defaultScheduler;
            System.Threading.Tasks.Task task = System.Threading.Tasks.Task.Factory.StartNew(() =>
            {
                try
                {
                    if (doWork != )
                    {
                        doWork();
                    }
                }
                catch (Exception ex)
                {
                    if (errorAction != ) errorAction(ex);
                    LogUtil.LogError(ex);
                }
            },CancellationToken.None,TaskCreationoptions.None,scheduler);
             task;
        }

        static System.Threading.Tasks.Task Run(Action doWork,1)">)
        {
            System.Threading.Tasks.Task task = System.Threading.Tasks.Task.Factory.StartNew(() =>) errorAction(ex);
                    LogUtil.LogError(ex);
                }
            });
             封装dispatcher.BeginInvoke 
         例：ThreadHelper.BeginInvoke(this.dispatcher,() => { },1)">void BeginInvoke(dispatcher dispatcher,Action action,1)">)
        {
            dispatcher.InvokeAsync(new Action(() =>
                {
                    DateTime dt = DateTime.Now;
                    action();
                    double d = DateTime.Now.Subtract(dt).TotalSeconds;
                    if (d > 0.01) LogUtil.Log(ThreadHelper.BeginInvoke UI耗时：" + d + 秒 " + action.Target.ToString());
                }
                ) errorAction(ex);
                    LogUtil.LogError(ex);
                }
            }),dispatcherPriority.Background);
        }
    }
}

View Code

测试方法：

 Test23()
{
    //变量定义
    DateTime dt = DateTime.Now;
    Random rnd =  Random();
    int taskCount = 1000;
    LimitedTaskScheduler scheduler =  LimitedTaskScheduler();

    生成测试数据
    BlockingCollection<double> _data = new BlockingCollection<double>();
    0; i < taskCount; i++)
    {
        _data.Add(rnd.NextDouble());
    }

    数据计算
    Thread thread = 
    {
        dt = DateTime.Now;
        )
        {
            ThreadHelper.Run(() =>
            {
                Thread.Sleep(50);
                double a;
                if (_data.TryTake( a))
                {
                    double r = Math.PI * a;
                }
            },scheduler);
        }
         DateTime.Now.Subtract(dt).TotalSeconds;

        this.BeginInvoke(
        {
            textBox1.Text += 调用" + taskCount + 次ThreadHelper.Run耗时：" + d.ToString() + 秒\r\n;
        }));
    }));
    thread.IsBackground = ;
    thread.Start();

    数据计算耗时
    Thread thread2 = 
    {
        while (_data.Count > )
        {
            Thread.Sleep();
        }
        数据计算结束，耗时：;
        }));
    }));
    thread2.IsBackground = ;
    thread2.Start();

    scheduler.dispose();
}

private LimitedTaskScheduler _scheduler =  LimitedTaskScheduler();
 Test24()
{
    点击按钮耗时
    DateTime dt = DateTime.Now;
    ThreadHelper.Run(() => DateTime.Now.Subtract(dt).TotalSeconds;
        点击按钮耗时：;
        }));
    },_scheduler);
}

View Code

事件方法：

void button1_Click(object sender,EventArgs e)
{
    Test23();
}

void button2_Click(View Code

测试操作步骤：

依次点击3次button1和1次button2

使用Run测试结果：

使用Run2测试结果：

结论：使用Run，点击button2时，卡了好几秒才出来结果，而使用Run2，点击button2时，立即显示结果，button2的操作本身应该耗时极少。

现实意义：当一批耗时任务无脑使用Task.Factory.StartNew时，另一个使用Task.Factory.StartNew的任务就无法及时响应了。

C# Task 多任务 限制Task并发数量

相关推荐

C# Task 多任务限制Task并发数量