實現一個所有任務都是按各自自定義的時間間隔週期性執行的執行緒池

阿新 • • 發佈：2019-01-21

在最近的一個專案中，經常有非常多個數據報文需要按各自的特定週期傳送，而且使用者可以控制是否繼續傳送、傳送多少次、傳送的時間週期等，如果每個傳送作業都開一個執行緒，對機器的開銷會特別大，而如果使用C#自帶的執行緒池就不能很好的實現這些功能。基於此，需要自己實現一個簡單的執行緒池。

下面這個自定義執行緒池的特點是，可執行多個週期性作業，每個作業可以由使用者自己控制是否繼續執行、加入或離開執行緒池、實時改變執行週期、執行次數、暫停執行和繼續執行、停止執行緒池和重啟執行緒池。並且對週期性作業的準時性做了優化。不足是暫時還未實現執行緒池的可伸縮性，執行緒數量是固定的。不過這隻需要做較小改動就可以實現。

實現如下：

1、週期性作業類

/// <summary>
    /// 時間作業的具體作業方法的委託
    /// </summary>
    public delegate void TaskFunc();

    /// <summary>
    /// 時間作業
    /// <note>成員包括剩餘做幾次、是否做、每次作業間隔多久、最後一次作業的時間滴答數</note>
    /// </summary>
    public class TimeTask
    {
        private int _TaskID = -1;
        private int _LeftCount = -1;            //剩餘做幾次，-1表示一直做
        private bool _IsDoing = true;                //是否執行任務的標記
        private bool _IsInPool = true;                //是否在池中的標記
        private long _Interval = 20000000;      //時間間隔 毫微秒（10^-4ms）
        private long _NextRunStartTime = 0;          //下一次計劃作業時間滴答數
        private long _RanCount = 0;                 //作業已執行次數
        private long _AverageRunTime = 0;           //作業平均執行時間
        private TaskFunc _Func;                 //作業方法

        public int TaskID { get { return _TaskID; } set { _TaskID = value; } }
        public int LeftCount { get { return _LeftCount; } set { _LeftCount = value; } }
        public bool IsDoing { get { return _IsDoing; } set { _IsDoing = value; } }
        public bool IsInPool { get { return _IsInPool; }  set { _IsInPool = value; } }
        public long Interval { get { return _Interval; } set { _Interval = value; } }
        public long NextRunStartTime { get { return _NextRunStartTime; } }
        public long AverageRunTime { get { return _AverageRunTime; } }
        public TaskFunc Func
        {
            get { return _Func; }
            set { _Func = value; }
        }
        //執行作業
        public void run()
        {
            if (_LeftCount != 0 && _Func != null)
            {
                long _LastRunStartTime = DateTime.Now.Ticks;
                _Func();
                long _LastRunFinishTime = DateTime.Now.Ticks;
                long _NowRunTime = (_LastRunFinishTime - _LastRunStartTime);
                _NextRunStartTime = _LastRunFinishTime + _Interval - _NowRunTime;
                _AverageRunTime = (_AverageRunTime * _RanCount + _NowRunTime) / (++_RanCount);
                //System.Console.Out.WriteLine("當前耗時： " + _NowRunTime / 10000 + "ms");
                //System.Console.Out.WriteLine("平均耗時： " + _AverageRunTime / 10000 + "ms");
                if (_LeftCount > 0)
                    _LeftCount--;
            }   
        }
        //從執行緒池中拆卸此作業
        public void dropFromPool()
        {
            _IsInPool = false;
        }
        //停止執行此任務
        public void stop()
        {
            _IsDoing = false;
        }
        //重新啟動此任務
        public void restart()
        {
            _IsDoing = true;
        }
        //過載equal
        public override bool Equals(object obj)
        {
            try
            {
                TimeTask _task = (TimeTask)obj;
                return _TaskID == _task.TaskID;
            }
            catch (Exception) { return false; }
        }
    }

2、自定義執行緒池類

public class MyThreadPool
    {
        private static int _WORKNUM = 5;                                                //執行緒池執行緒數
        private static long _LITTLETIMESPAN = 50000;                                    //小時間間隔 毫微秒（10^-4ms）
        private static int _TASKOFFSETFACTOR = 3;                                       //作業排隊因子
        private static List<TimeTask> _TaskList = new List<TimeTask>();                 //作業佇列
        private static WorkThread[] _WorksThread = new WorkThread[_WORKNUM];            //執行緒池中的所有執行緒
        private static bool _isRun = true;                                              //控制執行緒池執行與否的標誌
        private static int _NextTaskID = 0;                                             //下一個作業的ID

        #region 獲取單例
        private static readonly MyThreadPool _instance = new MyThreadPool();
        static MyThreadPool()
        {
            _isRun = true;
            for (int _i = 0; _i < _WORKNUM; _i++)
            {
                _WorksThread[_i] = new WorkThread();
                _WorksThread[_i].start();
            }
        }
        private MyThreadPool()
        {
            _isRun = true;
            for (int _i = 0; _i < _WORKNUM; _i++)
            {
                _WorksThread[_i] = new WorkThread();
                _WorksThread[_i].start();
            }
        }
        public static MyThreadPool Instance
        {
            get
            {
                return _instance;
            }
        }
        #endregion

        //增加作業，並提供作業ID
        public void addTask(TimeTask _task)
        {
            //作業需要先執行一次
            _task.run();
            _task.TaskID = _NextTaskID++;
            lock (_TaskList)
            {
                _TaskList.Add(_task);
            }
        }
        //開啟執行緒池中所有執行緒
        public void startAllThread()
        {
            if (!_isRun)
            {
                _isRun = true;
                for (int _i = 0; _i < _WORKNUM; _i++)
                {
                    _WorksThread[_i].start();
                }
            }
        }
        //自然停止所有執行緒（等待所有作業單個執行完畢）
        public void stopAllThread()
        {
            _isRun = false;
        }
        //清空作業列表
        public void clearTaskList()
        {
            lock (_TaskList)
            {
                _TaskList.Clear();
            }
        }
        //刪除某一項作業
        public bool removeTask(int _TaskID)
        {
            lock (_TaskList)
            {
                foreach(TimeTask _iter in _TaskList)
                {
                    if (_iter.TaskID == _TaskID)
                        return _TaskList.Remove(_iter);
                }
            }
            return false;
        }


        /// <summary>
        /// 執行緒池中的單個執行緒
        /// <note>MyThreadPool的內部類</note>
        /// </summary>
        private class WorkThread
        {
            //執行緒實體
            private Thread _thisThread; 
                                                            
            public WorkThread()
            {
            }
            //開啟執行緒
            public void start()
            {
                if (_thisThread == null)
                {
                    _thisThread = new Thread(_job);
                    _thisThread.Start();
                    return;
                }
                if (_thisThread.IsAlive) { }
                else
                {
                    _thisThread = new Thread(_job);
                    _thisThread.Start();
                    return;
                }
            }
            //執行緒執行方法：取出作業並執行，執行完畢後再次放回作業
            private void _job()
            {
                //判斷執行緒池是否要執行
                while (_isRun)
                {
                    TimeTask _nowTask = null;
                    //取出列表中第一個任務
                    lock (_TaskList)
                    {
                        if (_TaskList.Count != 0)
                        {
                            _nowTask = _TaskList.First();
                            _TaskList.RemoveAt(0);
                        }
                    }
                    //如果作業佇列為空，則掛起執行緒100ms後重新取作業
                    if (_nowTask == null) { Thread.Sleep((int)(_LITTLETIMESPAN/10000));continue; }
                    else
                    {
                        //拆卸作業，取出作業不往佇列中再次新增
                        if (!_nowTask.IsInPool)
                        {
                            //
                        }
                        //剩餘作業次數不為0且作業標記為執行
                        else if (_nowTask.LeftCount != 0 && _nowTask.IsDoing)
                        {
                            //如果當前時間小於計劃執行時間的負誤差，就將作業排到佇列末尾
                            if (DateTime.Now.Ticks < _nowTask.NextRunStartTime - _LITTLETIMESPAN)
                            {
                                lock (_TaskList)
                                {
                                    _TaskList.Add(_nowTask);
                                }
                            }
                            //如果當前時間大於等於計劃執行時間，就執行作業，然後將作業排到佇列末尾
                            else if (DateTime.Now.Ticks >= _nowTask.NextRunStartTime)
                            {
                                _nowTask.run();
                                lock (_TaskList)
                                {
                                    _TaskList.Add(_nowTask);
                                }
                            }
                            //如果當前時間處於計劃執行時間的負誤差之中，就將作業排到佇列中靠前的位置（取決於_TASKOFFSETFACTOR）
                            else
                            {
                                lock (_TaskList)
                                {
                                    if (_TaskList.Count < _WORKNUM)
                                    {
                                        _TaskList.Add(_nowTask);
                                    }
                                    else
                                    {
                                        _TaskList.Insert(_TaskList.Count / _TASKOFFSETFACTOR, _nowTask);
                                    }
                                }
                            }
                        }
                        //如果當前作業剩餘執行次數為零或者作業標記為不執行，就將此作業排到佇列末尾
                        else
                        {
                            lock (_TaskList)
                            {
                                _TaskList.Add(_nowTask);
                            }
                        }
                    }
                }
            }
        }

    }

3、測試結果總結：

編寫了一個小Demo對此執行緒池進行了測試，測試使用了10個週期性作業、每個作業的執行週期和執行時間都不相同。

以下是測試資料：（資料項從左至右依次為作業開始執行時間、作業名稱、作業模擬執行時間【呼叫sleep】，作業名稱的數字位也代表執行週期，如Task5代表此作業5秒執行一次）

12:23:06:903: Task1 --> 50ms
12:23:06:906: Task2 --> 20ms
12:23:06:908: Task3 --> 500ms
12:23:06:909: Task4 --> 200ms
12:23:06:911: Task5 --> 10ms
12:23:06:912: Task6 --> 30ms
12:23:06:914: Task7 --> 100ms
12:23:06:916: Task8 --> 40ms
12:23:06:918: Task9 --> 60ms
12:23:06:920: Task10 --> 70ms
12:23:08:66: Task1 --> 50ms
12:23:09:67: Task1 --> 50ms
12:23:09:155: Task2 --> 20ms
12:23:10:312: Task1 --> 50ms
12:23:10:348: Task3 --> 500ms
12:23:11:235: Task4 --> 200ms
12:23:11:312: Task1 --> 50ms
12:23:11:486: Task2 --> 20ms
12:23:12:38: Task5 --> 10ms
12:23:12:312: Task1 --> 50ms
12:23:12:986: Task6 --> 30ms
12:23:13:410: Task1 --> 50ms
12:23:13:950: Task3 --> 500ms
12:23:14:386: Task7 --> 100ms
12:23:14:409: Task1 --> 50ms
12:23:14:842: Task2 --> 20ms
12:23:15:231: Task8 --> 40ms
12:23:15:409: Task1 --> 50ms
12:23:15:429: Task4 --> 200ms
12:23:16:16: Task9 --> 60ms
12:23:16:480: Task1 --> 50ms
12:23:16:923: Task2 --> 20ms
12:23:16:950: Task3 --> 500ms
12:23:17:126: Task10 --> 70ms
12:23:17:415: Task5 --> 10ms
12:23:17:516: Task1 --> 50ms
12:23:18:750: Task1 --> 50ms
12:23:18:985: Task6 --> 30ms
12:23:19:490: Task2 --> 20ms
12:23:19:578: Task4 --> 200ms
12:23:19:988: Task3 --> 500ms

發現作業執行的週期準時性存在較小的誤差，但都保持在毫秒級，滿足當前專案的需求。

實現一個所有任務都是按各自自定義的時間間隔週期性執行的執行緒池

實現一個所有任務都是按各自自定義的時間間隔週期性執行的執行緒池

Android螢幕長按事件(自定義時間長短)

[MFC]自定義時間間隔Timer

quarz實現一個定時任務

【JAVA】簡單實現一個阻塞任務佇列

求助：Python菱形繼承傳參問題,是否應該使用super？每一個父類都有各自引數該如何傳參？以及使用了super之後父類還能否獨自生成物件？

net core天馬行空系列：一個介面多個實現類，利用mixin技術通過自定義服務名，實現精準屬性注入

實現一個“計劃任務”機制

JAVA實現延時過期MAP 支持自定義過期觸發事件

vue $emit子元件傳出多個引數，如何在父元件中在接收所有引數的同時新增自定義引數

h5 range實現slider滑塊功能及樣式自定義

讓你的app提升一個檔次－Android酷炫自定義控制元件

ssm整合shiro通過自定義Realm實現認證登入、許可權處理、自定義role攔截、MD5加密

Shiro——實現許可權控制demo思路（包含自定義標籤hasAnyPermission）

實現按鈕級許可權管理--使用jsp自定義標籤

java Collections.sort()實現List排序的預設方法和自定義方法

Android自定義時間軸的實現

Android構建一個通用的WebView(二)：自定義的錯誤頁面、快取資料，離線瀏覽

在所有WordPress文章底部新增自定義內容

Spring Boot實現OAuth 2.0（二）-- 自定義許可權驗證

實現一個所有任務都是按各自自定義的時間間隔週期性執行的執行緒池

相關推薦