自制线程池4

需求：

有一种任务需要定时的执行，而且非常的耗时，因此我把它放到线程池中执行，并设置线程池为1，如果该任务已经在队列中或正在执行该任务，则不要再将该任务加入线程池中了。

测试代码如下

<!--<br /> <br /> Code highlighting produced by Actipro CodeHighlighter (freeware)<br /> http://www.CodeHighlighter.com/<br /> <br /> --> 1 using System;
 2 using System.Collections.Generic;
 3 using System.Linq;
 4 using System.Text;
 5 using System.Threading;
 6 using ThreadPool2;
 7 
 8 namespace ThreadPoolTest.MyThreadPool2Test
 9 {
10     class Class6
11     {
12         static void Main(string[] args)
13         {
14             MyThreadPool2 pool=new MyThreadPool2(1,true,30000);
15             object obj=new object();
16             Random rnd=new Random();
17             for (var i = 0; i < 20;i++ )
18                 pool.QueueUserWorkItem(call, obj, rnd.Next(1,4).ToString(), succ, err);
19             Console.ReadLine();
20         }
21 
22         private static void err(object state)
23         {
24                 Console.WriteLine("err");
25         }
26 
27         private static void succ(object state, object result)
28         {
29             Console.WriteLine("succ");
30         }
31 
32         private static object call(object state)
33         {
34             while(true)
35             {
36                 Thread.Sleep(2000);
37                 Console.WriteLine("exec");
38             }
39         }
40     }
41 }
42 

 

线程池代码如下，

 

<!--<br /> <br /> Code highlighting produced by Actipro CodeHighlighter (freeware)<br /> http://www.CodeHighlighter.com/<br /> <br /> -->using System;
using System.Collections;
using System.Collections.Generic;
using System.Diagnostics;
using System.Linq;
using System.Text;
using System.Threading;
using Amib.Threading.Internal;
using Rhino.Commons;

namespace ThreadPool2
{
    public delegate object WaitCallback2(object state);
    public delegate void SuccCallback(object state, object result);
    public delegate void ErrCallback(object state);
    /**//// <summary>
    /// 此线程池的作用是将某一类特殊的任务交给此线程池执行，
    /// 可以设定该线程池的最大线程数，
    /// 这类线程池的优点时，占用的资源少，优先级低，
    /// 适合于执行任务需要长期执行，不考虑时间因素的任务
    /// 同时根据在传入线程池时的标记key，可以Aborted指定任务，
    /// 若该任务正在执行或尚在执行队列中
    /// </summary>
    public class MyThreadPool2
    {
        /**//// <summary>
        /// 任务执行队列
        /// </summary>
        //static ThreadSafeQueue<WorkerThread> queue = new ThreadSafeQueue<WorkerThread>();
        List<WorkerThread> queue = new List<WorkerThread>();
        /**//// <summary>
        /// 目前暂定为只使用一个线程，以免耗近资源
        /// </summary>
        SynchronizedDictionary<string, WorkerThread> dict = new SynchronizedDictionary<string, WorkerThread>();
        private object state;
        AutoResetEvent wait = new AutoResetEvent(false);
        AutoResetEvent wait2 = new AutoResetEvent(false);
        private int MaxLimitedTime { get; set; }
        private bool IsLimitedExecTime { get; set; }
        private int IdleTimeout { get; set; }
        //private static int _maxThreadNum = 1;
        private int MaxThreadNum
        {
            //get { return _maxThreadNum; }
            //set { _maxThreadNum = value; }
            get;
            set;
        }
        private MyThreadPool2()
        {
            //System.Threading.ThreadPool.RegisterWaitForSingleObject(wait, new WaitOrTimerCallback(aa), state, 2000,true);
            //SetMaxThreadNum(2);
            //SetMaxExecTime(false, 10000);
        }
        /**//// <summary>
        /// 设置专用线程池的初始参数
        /// </summary>
        /// <param name="num">线程池的最大线程数，最小为１</param>
        /// <param name="b">是否起用限制最大单个任务执行时间设定</param>
        /// <param name="MaxLimitedTime">单个任务执行的最大时间</param>
        public MyThreadPool2(int num, bool b, int MaxLimitedTime)
        {
            System.Threading.ThreadPool.RegisterWaitForSingleObject(wait, new WaitOrTimerCallback(aa), state, 2000, true);
            if (num < 1)
                num = 1;
            MaxThreadNum = num;
            IsLimitedExecTime = b;
            this.MaxLimitedTime = MaxLimitedTime;
            if (IsLimitedExecTime)
                System.Threading.ThreadPool.RegisterWaitForSingleObject(wait2, new WaitOrTimerCallback(bb), state,
                                                                        this.MaxLimitedTime, true);
        }

        /**//// <summary>
        /// 定时将队列中的数据装载到线程中执行，如果还没有到达最大线程数还有任务则创建线程
        /// </summary>
        /// <param name="state"></param>
        /// <param name="timedOut"></param>
        private void aa(object state, bool timedOut)
        {
            //Console.WriteLine("执行aa()将队列中的任务加到线程中");
            lock(WorkerThread.Manual){
            WorkerThread.Manual.Reset();
            lock (queue)
            {
                Console.WriteLine("queue count={0}",queue.Count);
                //判断任务队列中有无积压的任务且有无空闲的线程，如果符合上述条件则执行之
                List<string> removeKey = new List<string>();
                List<WorkerThread> newTask = new List<WorkerThread>();
                List<string> tasks = new List<string>();
                //Dictionary<string,WorkerThread> addDict=new Dictionary<string, WorkerThread>();
                foreach (var kvp in dict)
                {//kvp.Value.ThreadState == ThreadState.Unstarted || 
                    //if (kvp.Value.Thread.ThreadState == ThreadState.Suspended)

                    //将不活动的线程记录下来并移除
                    if (!kvp.Value.Thread.IsAlive)
                        tasks.Add(kvp.Key);
                    //将活动且空闲的线程赋于新的任务
                    if (kvp.Value.Thread.IsAlive == true && kvp.Value.CurrentThreadState == WorkerThreadState.Idle)
                    {
                        //dict.Remove(kvp.Key);//cancle because of lock

                        WorkerThread a = queue.FirstOrDefault();
                        if (a != null)
                        {
                            removeKey.Add(kvp.Key);
                            //addDict.Add(a.Key, kvp.Value.Change(a));
                            newTask.Add(kvp.Value.Change(a));
                            //a.Thread = kvp.Value.Thread;
                            //newTask.Add(a);
                            queue.RemoveAt(0);
                            //dict.Add(a.Key, kvp.Value.Change(a));//cancle because of lock
                            //将参数加到线程中，并改变线程的状态
                            //dict[a.Key].Thread.Resume();
                        }
                        else
                            break;
                        //else
                        //{
                        //    System.Threading.ThreadPool.RegisterWaitForSingleObject(wait, new WaitOrTimerCallback(aa), state,
                        //                                                            2000, true);
                        //    return;
                        //}

                    }
                }
                tasks.ForEach(t => 
                { 
                    dict.Remove(t);
                    Debug.WriteLine("移除销毁线程对应的dict中的键值项，key="+t);
                });
                removeKey.ForEach(t => dict.Remove(t));
                newTask.ForEach(t =>
                {
                    Debug.WriteLine("复用线程用于执行新任务"+t.Key);
                    dict.Add(t.Key, t);
                    //t.StartExecTime = DateTime.Now;
                    t.Auto.Set();
                    //t.CurrentThreadState = WorkerThreadState.Busy;
                    //t.Thread.Resume();
                });
                while (queue.Count > 0 && dict.Count < MaxThreadNum)
                {
                    //未到线程池最大池程数时，增加线程
                    WorkerThread b = queue.FirstOrDefault();
                    if (b != null)
                    {
                        queue.RemoveAt(0);
                        //Thread thd = new Thread(new ThreadStart(b.Exec));
                        //thd.Priority = ThreadPriority.Lowest;
                        //dict.Add(b.Key, thd);
                        //thd.Start();
                        WorkerThread wt = new WorkerThread();
                        wt.Start(b);
                        dict.Add(wt.Key, wt);
                        wt.Thread.Start();
                        Debug.WriteLine("新建线程用于执行新任务"+ wt.Key);


                        //将参数加到线程中，并改变线程的状态
                    }


                }
                System.Threading.ThreadPool.RegisterWaitForSingleObject(wait, new WaitOrTimerCallback(aa), state, 2000,
http://www.cnblogs.com/Imag

分享到：

分布式信息采集程序preview | 目前来说比较.net下最好的bdb操作封装（附 ...

• 2008-08-24 22:28
• 浏览 697
• 评论(0)
• 查看更多