草庐IT

C# 简单实现线程池

ncqingchuan 2023-03-28 原文

NET 6 环境开发 实现 线程数量,任务队列,非核心线程,及核心线程活跃时间的管理。

namespace CustomThreadPool;
/// <summary>
/// 线程池类
/// </summary>
public class ThreadPoolExecutor
{
    /// <summary>
    /// 核心线程的任务队列
    /// </summary>
    private readonly Queue<WorkTask> tasks = new Queue<WorkTask>();
    /// <summary>
    /// 最大核心线程数
    /// </summary>
    private int coreThreadCount;
    /// <summary>
    /// 最大非核心线程数
    /// </summary>
    private int noneCoreThreadCount;
    /// <summary>
    /// 当前运行的核心线程的数量
    /// </summary>
    private int runCoreThreadCount;
    /// <summary>
    /// 当前运行的非核心线程的数量
    /// </summary>
    private int runNoneCoreThreadCount;
    /// <summary>
    /// 核心线程队列的最大数
    /// </summary>
    private int maxQueueCount;
    /// <summary>
    /// 当核心线程空闲时最大活跃时间
    /// </summary>
    private int keepAliveTimeout;
    /// <summary>
    /// 设置是否为后台线程
    /// </summary>
    private bool isBackground;
    private ThreadPoolExecutor() { }
    /// <summary>
    /// 
    /// </summary>
    /// <param name="CoreThreadCount">核心线程数</param>
    /// <param name="TotalThreadCount">总线程数</param>
    /// <param name="IsBackground">是否为后台线程</param>
    /// <param name="QueueCount">核心队列的最大数</param>
    /// <param name="KeepAliveTimeout">当核心线程空闲时最大活跃时间</param>
    /// <exception cref="ArgumentOutOfRangeException"></exception>
    /// <exception cref="ArgumentException"></exception>
    public ThreadPoolExecutor(int CoreThreadCount = 5, int TotalThreadCount = 10, bool IsBackground = true, int QueueCount = 200, int KeepAliveTimeout = 0)
    {
        if (CoreThreadCount < 1) throw new ArgumentOutOfRangeException(nameof(CoreThreadCount), CoreThreadCount, null);
        if (TotalThreadCount < CoreThreadCount) throw new ArgumentException($"{nameof(TotalThreadCount)}:{TotalThreadCount} must be greater than {nameof(CoreThreadCount)}:{CoreThreadCount}");
        if (QueueCount < 0) throw new ArgumentOutOfRangeException(nameof(QueueCount), QueueCount, null);
        if (KeepAliveTimeout < 0) throw new ArgumentOutOfRangeException(nameof(KeepAliveTimeout), KeepAliveTimeout, null);
        coreThreadCount = CoreThreadCount;
        noneCoreThreadCount = TotalThreadCount - CoreThreadCount;
        keepAliveTimeout = KeepAliveTimeout;
        maxQueueCount = QueueCount;
        isBackground = IsBackground;
    }
    /// <summary>
    /// 执行任务
    /// </summary>
    /// <param name="task">一个自定义任务</param>
    /// <exception cref="ArgumentNullException">任务为null时,抛出该错误</exception>
    /// <exception cref="NotSupportedException">当核心任务队列已满且非核心线程最大数为0时抛出该错误</exception>
    public void QueueTask(WorkTask task)
    {
        if (task == null) throw new ArgumentNullException(nameof(task));
        lock (tasks)
        {
            tasks.Enqueue(task);
            if (tasks.Count <= maxQueueCount)
            {
                if (runCoreThreadCount < coreThreadCount)
                {
                    ++runCoreThreadCount;
                    Run(true);
                }
            }
            else
            {
                if (noneCoreThreadCount > 0 && runNoneCoreThreadCount < noneCoreThreadCount)
                {
                    ++runNoneCoreThreadCount;
                    Run(false);
                }
            }
        }
    }

    private void Run(bool isCore)
    {
        Tuple<int, bool> state = new(keepAliveTimeout, isCore);
        Thread thread = new(t => Excute(t))
        {
            Name = Guid.NewGuid().ToString("D"),
            IsBackground = isBackground
        };
        thread.Start(state);
    }

    private void Excute(object? state)
    {
        if (state == null) return;
        var parameter = (Tuple<int, bool>)state;
        bool first = true;
        DateTime firstTime = DateTime.Now;
        while (true)
        {
            WorkTask? item = null;
            lock (tasks)
            {
                if (tasks.Count > 0)
                {
                    first = true;
                    item = tasks.Dequeue();
                }
                else
                {
                    if (parameter.Item2)
                    {

                        if (first)
                        {
                            firstTime = DateTime.Now;
                            first = false;
                        }
                        if ((DateTime.Now - firstTime).TotalMilliseconds > parameter.Item1)
                        {
                            --runCoreThreadCount;
                            break;
                        }
                    }
                    else
                    {
                        --runNoneCoreThreadCount;
                        break;
                    }
                }
            }
            item?.Runsynchronous();
        }
    }
}
namespace CustomThreadPool;
/// <summary>
/// 包装的任务类
/// </summary>
public class WorkTask
{
    public static WorkTaskFactory Factory { get; private set; } = new WorkTaskFactory();
    /// <summary>
    /// 任务运行结束时触发该事件
    /// </summary>
    public event Action<WorkTask>? TaskCompleted;
    /// <summary>
    /// 任务ID
    /// </summary>
    private static int _id = 0;
    /// <summary>
    /// 委托给任务不带执行参数的代码
    /// </summary>
    private readonly Action? action;
    /// <summary>
    /// 委托给任务执行的带输入参数代码
    /// </summary>
    private readonly Action<object?>? actionWithParamter;
    /// <summary>
    /// 线程间的同步事件
    /// </summary>
    public AutoResetEvent WaitHandle { get; protected set; } = new AutoResetEvent(false);
    /// <summary>
    /// 执行代码的参数
    /// </summary>
    public object? State { get; protected set; }
    /// <summary>
    /// 接收任务抛出的异常
    /// </summary>
    public WorkTaskException? Exception { get; protected set; }
    /// <summary>
    /// 任务是否完成标志
    /// </summary>
    public bool IsCompleted { get; protected set; } = false;
    /// <summary>
    /// 任务知否有异常
    /// </summary>
    public bool IsFaulted { get; protected set; } = false;
    /// <summary>
    /// 任务状态
    /// </summary>
    public WorkTaskStatus Status { get; protected set; } = WorkTaskStatus.Created;
    public int Id { get { return Interlocked.Increment(ref _id); } }
    protected WorkTask() { }

    protected void OnTaskCompleted(WorkTask sender)
    {
        TaskCompleted?.Invoke(sender);
    }
    public WorkTask(Action action)
    {
        this.action = action ?? throw new ArgumentNullException(nameof(action));
    }
    public WorkTask(Action<object?> action, object state)
    {
        actionWithParamter = action ?? throw new ArgumentNullException(nameof(action));
        this.State = state;
    }
    /// <summary>
    /// 任务的同步方法
    /// </summary>
    public virtual void Runsynchronous()
    {
        if (Status != WorkTaskStatus.Created) return;
        Status = WorkTaskStatus.Running;
        try
        {
            action?.Invoke();
            actionWithParamter?.Invoke(State);
        }
        catch (Exception ex)
        {
            Exception = new WorkTaskException(ex.Message, ex);
            IsFaulted = true;
        }
        finally
        {
            OnTaskCompleted(this);
            WaitHandle.Set();
            IsCompleted = true;
            Status = WorkTaskStatus.RanToCompleted;
        }
    }
    /// <summary>
    /// 通过调用线程执行的方法
    /// </summary>
    public void Start()
    {
        Factory.ThreadPoolExcutor?.QueueTask(this);
    }
    /// <summary>
    /// 通过调用线程执行的方法
    /// </summary>
    /// <param name="executor">线程池管理类</param>
    public void Start(ThreadPoolExecutor executor)
    {
        executor.QueueTask(this);
    }
    /// <summary>
    /// 执行一组任务并等待所有任务完成。
    /// </summary>
    /// <param name="tasks">一组任务</param>
    /// <returns>所有任务是否都接收到完成的信号。</returns>
    public static bool WaitAll(WorkTask[] tasks)
    {
        var result = true;
        foreach (var task in tasks)
        {
            result = result && task.WaitHandle.WaitOne();
        }
        return result;
    }

    /// <summary>
    /// 执行一组任务并等待任意一个任务完成。
    /// </summary>
    /// <param name="tasks">一组任务</param>
    /// <returns>返回已完成任务的索引</returns>
    public static int WaitAny(WorkTask[] tasks)
    {
        var index = new Random().Next(0, tasks.Length - 1);
        tasks[index].WaitHandle.WaitOne();
        return index;
    }
}
/// <summary>
/// 具有返回类型的任务
/// </summary>
/// <typeparam name="TResult"></typeparam>
public class WorkTask<TResult> : WorkTask
{
    private readonly Func<TResult>? func;
    private readonly Func<object?, TResult>? funcWithParameter;

    protected TResult? _result = default(TResult);
    public TResult? Result
    {
        get
        {
            if (!isSetSignal)
                WaitHandle.WaitOne();
            return _result;
        }
    }
    public WorkTask(Func<TResult> func)
    {
        this.func = func ?? throw new ArgumentNullException(nameof(func));
    }
    public WorkTask(Func<object?, TResult> func, object? state)
    {
        this.funcWithParameter = func ?? throw new ArgumentNullException(nameof(func));
        this.State = state;
    }

    private bool isSetSignal = false;
    public override void Runsynchronous()
    {
        if (Status != WorkTaskStatus.Created) return;
        Status = WorkTaskStatus.Running;
        try
        {
            if (func != null) _result = func();
            if (funcWithParameter != null) _result = funcWithParameter(State);
        }
        catch (Exception ex)
        {
            Exception = new WorkTaskException(ex.Message, ex);
            IsFaulted = true;
        }
        finally
        {
            OnTaskCompleted(this);
            isSetSignal = WaitHandle.Set();
            Status = WorkTaskStatus.RanToCompleted;
            IsCompleted = true;

        }
    }
}
public class WorkTaskException : Exception
{
    public WorkTaskException()
    {
    }
    public WorkTaskException(string Message)
        : base(Message)
    {
    }

    public WorkTaskException(string Message, Exception InnerException)
        : base(Message, InnerException)
    {
    }
}

public enum WorkTaskStatus
{
    /// <summary>
    /// 已创建
    /// </summary>
    Created = 0,
    /// <summary>
    /// 正在运行
    /// </summary>
    Running = 1,
    /// <summary>
    /// 已完成
    /// </summary>
    RanToCompleted = 2,
}
namespace CustomThreadPool;

public class WorkTaskFactory
{
    public ThreadPoolExecutor? ThreadPoolExcutor { get; private set; }
    public WorkTaskFactory(ThreadPoolExecutor excutor)
    {
        ThreadPoolExcutor = excutor;
    }

    public WorkTaskFactory()
    : this(new ThreadPoolExecutor(5, 10))
    {
    }

    public WorkTask StartNew(Action action, ThreadPoolExecutor? executor = null)
    {
        WorkTask task = new WorkTask(action);
        ThreadPoolExcutor = executor ?? ThreadPoolExcutor;
        ThreadPoolExcutor?.QueueTask(task);
        return task;
    }

    public WorkTask<TResult> StartNew<TResult>(Func<object?, TResult> func, object? state, ThreadPoolExecutor? executor = null)
    {
        WorkTask<TResult> task = new WorkTask<TResult>(func, state);
        ThreadPoolExcutor = executor ?? ThreadPoolExcutor;
        ThreadPoolExcutor?.QueueTask(task);
        return task;
    }
}
namespace CustomThreadPool;
using System.Threading;
using System.Text;
using System;
using System.Diagnostics;
using System.Reflection.Emit;

class Program
{
    static void Main(string[] args)
    {
        int count = 5;
        ThreadPoolExecutor poolExcutor = new(5, 6, QueueCount: 5, KeepAliveTimeout: 2000);
        WorkTask<int?>[] workTasks = new WorkTask<int?>[count];
        for (int i = 0; i < count; i++) workTasks[i] = WorkTask.Factory.StartNew(t => Action(t), state: i, executor: poolExcutor);

        WorkTask<int> task = WorkTask.Factory.StartNew(t =>
        {
            Thread.Sleep(100);
            Console.WriteLine("start thread");
            return 100;
        }, state: null, executor: poolExcutor);

        Console.WriteLine("start main");
        WorkTask.WaitAll(workTasks);
        Console.WriteLine(task.Result);
        Console.WriteLine(workTasks.Sum(t => t.Result));
    }
    private static int? Action(object? t)
    {
        Thread.Sleep(2000);
        Console.WriteLine($"Task Id:{Environment.CurrentManagedThreadId},Parameter:{t}");
        return t == null ? default(int?) : (int)t + 1;
    }
}

调用结果

 

有关C# 简单实现线程池的更多相关文章

  1. ruby - RuntimeError(自动加载常量 Apps 多线程时检测到循环依赖 - 2

    我收到这个错误:RuntimeError(自动加载常量Apps时检测到循环依赖当我使用多线程时。下面是我的代码。为什么会这样?我尝试多线程的原因是因为我正在编写一个HTML抓取应用程序。对Nokogiri::HTML(open())的调用是一个同步阻塞调用,需要1秒才能返回,我有100,000多个页面要访问,所以我试图运行多个线程来解决这个问题。有更好的方法吗?classToolsController0)app.website=array.join(',')putsapp.websiteelseapp.website="NONE"endapp.saveapps=Apps.order("

  2. ruby - 如何根据特征实现 FactoryGirl 的条件行为 - 2

    我有一个用户工厂。我希望默认情况下确认用户。但是鉴于unconfirmed特征,我不希望它们被确认。虽然我有一个基于实现细节而不是抽象的工作实现,但我想知道如何正确地做到这一点。factory:userdoafter(:create)do|user,evaluator|#unwantedimplementationdetailshereunlessFactoryGirl.factories[:user].defined_traits.map(&:name).include?(:unconfirmed)user.confirm!endendtrait:unconfirmeddoenden

  3. ruby - 简单获取法拉第超时 - 2

    有没有办法在这个简单的get方法中添加超时选项?我正在使用法拉第3.3。Faraday.get(url)四处寻找,我只能先发起连接后应用超时选项,然后应用超时选项。或者有什么简单的方法?这就是我现在正在做的:conn=Faraday.newresponse=conn.getdo|req|req.urlurlreq.options.timeout=2#2secondsend 最佳答案 试试这个:conn=Faraday.newdo|conn|conn.options.timeout=20endresponse=conn.get(url

  4. c# - 如何在 ruby​​ 中调用 C# dll? - 2

    如何在ruby​​中调用C#dll? 最佳答案 我能想到几种可能性:为您的DLL编写(或找人编写)一个COM包装器,如果它还没有,则使用Ruby的WIN32OLE库来调用它;看看RubyCLR,其中一位作者是JohnLam,他继续在Microsoft从事IronRuby方面的工作。(估计不会再维护了,可能不支持.Net2.0以上的版本);正如其他地方已经提到的,看看使用IronRuby,如果这是您的技术选择。有一个主题是here.请注意,最后一篇文章实际上来自JohnLam(看起来像是2009年3月),他似乎很自在地断言RubyCL

  5. ruby - 用 Ruby 编写一个简单的网络服务器 - 2

    我想在Ruby中创建一个用于开发目的的极其简单的Web服务器(不,不想使用现成的解决方案)。代码如下:#!/usr/bin/rubyrequire'socket'server=TCPServer.new('127.0.0.1',8080)whileconnection=server.acceptheaders=[]length=0whileline=connection.getsheaders想法是从命令行运行这个脚本,提供另一个脚本,它将在其标准输入上获取请求,并在其标准输出上返回完整的响应。到目前为止一切顺利,但事实证明这真的很脆弱,因为它在第二个请求上中断并出现错误:/usr/b

  6. C# 到 Ruby sha1 base64 编码 - 2

    我正在尝试在Ruby中复制Convert.ToBase64String()行为。这是我的C#代码:varsha1=newSHA1CryptoServiceProvider();varpasswordBytes=Encoding.UTF8.GetBytes("password");varpasswordHash=sha1.ComputeHash(passwordBytes);returnConvert.ToBase64String(passwordHash);//returns"W6ph5Mm5Pz8GgiULbPgzG37mj9g="当我在Ruby中尝试同样的事情时,我得到了相同sha

  7. ruby-on-rails - 简单的 Ruby on Rails 问题——如何将评论附加到用户和文章? - 2

    我意识到这可能是一个非常基本的问题,但我现在已经花了几天时间回过头来解决这个问题,但出于某种原因,Google就是没有帮助我。(我认为部分问题在于我是一个初学者,我不知道该问什么......)我也看过O'Reilly的RubyCookbook和RailsAPI,但我仍然停留在这个问题上.我找到了一些关于多态关系的信息,但它似乎不是我需要的(尽管如果我错了请告诉我)。我正在尝试调整MichaelHartl'stutorial创建一个包含用户、文章和评论的博客应用程序(不使用脚手架)。我希望评论既属于用户又属于文章。我的主要问题是:我不知道如何将当前文章的ID放入评论Controller。

  8. ruby - 使用 Ruby 通过 Outlook 发送消息的最简单方法是什么? - 2

    我的工作要求我为某些测试自动生成电子邮件。我一直在四处寻找,但未能找到可以快速实现的合理解决方案。它需要在outlook而不是其他邮件服务器中,因为我们有一些奇怪的身份验证规则,我们需要保存草稿而不是仅仅发送邮件的选项。显然win32ole可以做到这一点,但我找不到任何相当简单的例子。 最佳答案 假设存储了Outlook凭据并且您设置为自动登录到Outlook,WIN32OLE可以很好地完成此操作:require'win32ole'outlook=WIN32OLE.new('Outlook.Application')message=

  9. 华为OD机试用Python实现 -【明明的随机数】 2023Q1A - 2

    华为OD机试题本篇题目:明明的随机数题目输入描述输出描述:示例1输入输出说明代码编写思路最近更新的博客华为od2023|什么是华为od,od薪资待遇,od机试题清单华为OD机试真题大全,用Python解华为机试题|机试宝典【华为OD机试】全流程解析+经验分享,题型分享,防作弊指南华为o

  10. 基于C#实现简易绘图工具【100010177】 - 2

    C#实现简易绘图工具一.引言实验目的:通过制作窗体应用程序(C#画图软件),熟悉基本的窗体设计过程以及控件设计,事件处理等,熟悉使用C#的winform窗体进行绘图的基本步骤,对于面向对象编程有更加深刻的体会.Tutorial任务设计一个具有基本功能的画图软件**·包括简单的新建文件,保存,重新绘图等功能**·实现一些基本图形的绘制,包括铅笔和基本形状等,学习橡皮工具的创建**·设计一个合理舒适的UI界面**注明:你可能需要先了解一些关于winform窗体应用程序绘图的基本知识,以及关于GDI+类和结构的知识二.实验环境Windows系统下的visualstudio2017C#窗体应用程序三.

随机推荐