C# 簡單實現線程池

来源:https://www.cnblogs.com/ncqingchuan/archive/2022/11/21/16913044.html
-Advertisement-
Play Games

NET 6 環境開發 實現 線程數量,任務隊列,非核心線程,及核心線程活躍時間的管理。 namespace CustomThreadPool; /// <summary> /// 線程池類 /// </summary> public class ThreadPoolExecutor { /// <s ...


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

調用結果

 


您的分享是我們最大的動力!

-Advertisement-
Play Games
更多相關文章
  • 來源:blog.csdn.net/u014454538/article/details/98515807 1. Java中的線程安全 Java線程安全:狹義地認為是多線程之間共用數據的訪問。 Java語言中各種操作共用的數據有5種類型:不可變、絕對線程安全、相對線程安全、線程相容、線程獨立 ① 不可 ...
  • 1、快速開始 1.1、現有一張 User 表,其表結構如下 | id | name | age | emali | | | | | | | 1 | Jone | 18 | [email protected] | | 2 | Jack | 20 | [email protected] | | 3 ...
  • 在JSP頁面實現驗證碼校驗文章中當時是使用的Servlet類來進行的驗證碼校驗,但是這種方式並不能即時校驗,在正常情況下都是直接在用戶輸入之後就進行校驗,這樣對用戶來說很方便的。 AJAX 即“Asynchronous Javascript And XML”(非同步 JavaScript 和 XML) ...
  • 抽象類:可以作為頂層基類,從高層次規範編程介面 1、在abc模塊中,最常見的抽象類有Iterable,我們可以用他判斷一個對象是不是可迭代對象 1 from collections.abc import Iterable 2 3 4 class Valley: 5 6 def __iter__(se ...
  • 流程式控制制主要作用於:用來判斷某個程式在不同條件下,執行不同邏輯。 按結構分類,流程式控制制可分為:順序結構、選擇結構、迴圈結構 順序結構:程式從上往下執行 選擇結構:根據程式在不同條件下,選擇執行不同的邏輯代碼 迴圈結構:在一定的條件下,重覆執行某段代碼 選擇結構 java選擇結構的語句有:if語句 i ...
  • 虛擬機 1.1 發展歷程 1.1.1 java往事 ​ Java誕生在一群懶惰、急躁而傲慢的程式天才之中。 ​ 1990年12月,Sun的工程師Patrick Naughton被當時糟糕的Sun C++工具折磨的快瘋了。他大聲抱怨,並威脅要離開Sun轉投當時在Steve Jobs領導之下的NeXT公 ...
  • 1、流式查詢 1、實體類 package com.wanqi.pojo; import java.util.Date; /** * @Description TODO * @Version 1.0.0 * @Date 2022/9/12 * @Author wandaren */ public cl ...
  • 最近在學習fastdfs時遇到了一個困擾我數小時的問題,結果很感人,特在此記錄下來,也是為了讓其他小伙伴們避坑 (這裡我的fastdfs運行環境是在docker中) 事情是這樣的,用APIfox測試fastdfs文件上傳的時候發現明明返回的成功的結果但是在瀏覽器中使用路徑卻訪問不到伺服器上的圖片,瀏 ...
一周排行
    -Advertisement-
    Play Games
  • 簡介 本文的初衷是希望幫助那些有其它平臺視覺演算法開發經驗的人能快速轉入Halcon平臺下,通過文中的示例開發者能快速瞭解一個Halcon項目開發的基本步驟,讓開發者能把精力完全集中到演算法的開發上面。 首先,你需要安裝Halcon,HALCON 18.11.0.1的安裝包會放在文章末尾。安裝包分開發和 ...
  • 本文是對Datawhale的動手學數據分析課程的學習總結,記錄了整體的學習過程、答案以及個人感想,代碼較為詳細。 ...
  • JZ7重建二叉樹 描述 給定節點數為 n 的二叉樹的前序遍歷和中序遍歷結果,請重建出該二叉樹並返回它的頭結點。 例如輸入前序遍歷序列{1,2,4,7,3,5,6,8}和中序遍歷序列{4,7,2,1,5,3,8,6} 提示: 1.vin.length == pre.length 2.pre 和 vin ...
  • 我們都知道在Java編程中多線程的同步使用synchronized關鍵字來標識,那麼這個關鍵字在JVM底層到底是如何實現的呢。 我們先來思考一下如果我們自己實現的一個鎖該怎麼做呢: 首先肯定要有個標記記錄對象是否已經上鎖,執行同步代碼之前判斷這個標誌,如果對象已經上鎖線程就阻塞等待鎖的釋放。 其次要 ...
  • 目錄 一.OpenGL 色階 1.Windows OpenGL ES 版本 2.Windows OpenGL 版本 二.OpenGL 色階 GLSL Shader 三.猜你喜歡 零基礎 OpenGL ES 學習路線推薦 : OpenGL ES 學習目錄 >> OpenGL ES 基礎 零基礎 Ope ...
  • 1. 查看Linux伺服器版本信息 # cat /etc/redhat-release CentOS Linux release 7.4.1708 (Core) 2. 禪道開源版安裝包下載 wget http://dl.cnezsoft.com/zentao/9.8.2/ZenTaoPMS.9.8. ...
  • Spring 框架可以為 Java 應用程式開發提供全面的基礎設施支持,它是現在非常流行的 Java 開源框架,對於一個 Java 開發人員來說,熟練掌握 Spring 是必不可少的。 ...
  • 前言 本篇是c++總結的第二篇,關於c++的對象模型,在構造、拷貝虛函數上重點分析,也包含了c++11class的新用法和特性,如有不當,還請指教! c++三大特性 訪問許可權 ​ 在c++中通過public、protected、private三個關鍵字來控製成員變數和成員函數的訪問許可權,它們分別表示 ...
  • 一.小結 1.使用二維數組來存儲表格 2.可以使用以下語法來聲明二維數組變數: 元素類型[ ] [ ]數組變數 3.可以使用以下語法來創建二維數組變數: new 元素類型 [行的個數][列的個數] 4.使用下麵的語法表示二維數組中的每個元素: 數組變數[行下標][列的個數] 5.可使用數組初始化語法 ...
  • typimg是一款為typora編輯器提供圖像自定義上傳服務的工具,該工具將在typora中輸入的網路圖片、本地圖片、剪貼板圖片/截圖上傳到博客園,支持在MacOS、Windiws、Linux三個平臺上運行。 ...