.net内置的threadpool对于加入执行队列的任务，或是正在执行的任务无法取消，这对于我的项目来说有问题，因此要自定义一个线程池。
我的项目中具体的应用情节如下，某一个操作会非常耗时（将网址插入bdb中），如果将其加入线程池中，很可能将线程池中的资源耗尽，因此我希望我可以定义一个maxThreadNum用来控制执行此在操作最大允许同时执行的线程数，同时设定线程等级为最低ThreadPriority.Lowest，我只要这个操作慢慢执行就行了。
代码还需要重构，高手请指点下，对锁的机制还是不太清楚，
Code
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading;
using Rhino.Commons;
namespace ThreadPool
{
public static class MyThreadPool
{
static object obj = new object();
static AutoResetEvent wait = new AutoResetEvent(false);
static MyThreadPool()
{
System.Threading.ThreadPool.RegisterWaitForSingleObject(wait, new WaitOrTimerCallback(aa), state, 2000,true);
SetMaxThreadNum(1);
}
private static void aa(object state, bool timedOut)
{
lock (obj)
{
//判断任务队列中有无积压的任务且有无空闲的线程，如果符合上述条件则执行之
List<string> removeKey = new List<string>();
List<WorkerThread> newTask=new List<WorkerThread>();
//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)
{
//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));
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;
//}
}
}
removeKey.ForEach(t=>dict.Remove(t));
newTask.ForEach(t =>
{
dict.Add(t.Key, t);
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();
//将参数加到线程中，并改变线程的状态
}
}
System.Threading.ThreadPool.RegisterWaitForSingleObject(wait, new WaitOrTimerCallback(aa), state, 2000, true);
}
}
//private static int _maxThreadNum = 1;
public static int MaxThreadNum
{
//get { return _maxThreadNum; }
//set { _maxThreadNum = value; }
get; set;
}
public static void SetMaxThreadNum(int num)
{
if (num < 1)
num = 1;
MaxThreadNum = num;
}
/// <summary>
/// 任务执行队列
/// </summary>
//static ThreadSafeQueue<WorkerThread> queue = new ThreadSafeQueue<WorkerThread>();
static List<WorkerThread> queue=new List<WorkerThread>();
/// <summary>
/// 目前暂定为只使用一个线程，以免耗近资源
/// </summary>
static Dictionary<string, WorkerThread> dict = new Dictionary<string, WorkerThread>(1);
private static object state;
public static void Aborted(string key)
{
lock (obj)
{
WorkerThread v;
if (dict.TryGetValue(key, out v))
{
v.Thread.Abort();
dict.Remove(key);
}
int index = queue.FindIndex(t => t.Key == key);
if (index>-1)
queue.RemoveAt(index);
wait.Set();
}
}
public static void QueueUserWorkItem(WaitCallback callback, object state, string key)
{
WorkerThread p = new WorkerThread()
{
Callback = callback,
State = state,
Key = key
};
//queue.Enqueue(p);
queue.Add(p);
wait.Set();
}
}
}
public class WorkerThread
{
public Thread Thread { get; set; }
public string Key { get; set; }
public WaitCallback Callback { get; set; }
public Object State { get; set; }
public void Exec()
{
while (true)
{
this.Callback(this.State);
this.Thread.Suspend();
}
}
public WorkerThread Change(WorkerThread wt)
{
this.Key = wt.Key;
this.Callback = wt.Callback;
this.State = wt.State;
return this;
}
public void Start(WorkerThread wt)
{
this.Change(wt);
this.Thread = new Thread(new ThreadStart(this.Exec));
this.Thread.Priority = ThreadPriority.Lowest;
}
//public void Start(WaitCallback callback,Object state)
//{
// this.Callback = callback;
// this.State = state;
// if(this.Thread==null){
// this.Thread = new Thread(new ThreadStart(this.Exec));
// this.Thread.Priority = ThreadPriority.Lowest;
// this.Thread.IsBackground = true;
// this.Thread.Start();
// return;
// }
// if(this.Thread.ThreadState==ThreadState.Suspended)
// {
// this.Thread.Resume();
// }
//}
}
测试代码
Code
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading;
namespace ThreadPoolTest
{
public class Class1
{
static ManualResetEvent wait=new ManualResetEvent(false);
static void Main(string[] args)
{
object state=new object();
ThreadPool.MyThreadPool.QueueUserWorkItem(new WaitCallback(test), state, "aa");
ThreadPool.MyThreadPool.QueueUserWorkItem(new WaitCallback(test2), state, "bb");
System.Threading.Thread.Sleep(10000);
wait.Set();
System.Threading.Thread.Sleep(10000);
Console.WriteLine("aborted aa");
ThreadPool.MyThreadPool.Aborted("aa");
ThreadPool.MyThreadPool.QueueUserWorkItem(new WaitCallback(test3), state, "cc");
System.Threading.Thread.Sleep(10000);
Console.WriteLine("aborted bb");
ThreadPool.MyThreadPool.Aborted("bb");
ThreadPool.MyThreadPool.QueueUserWorkItem(new WaitCallback(test4), state, "dd");
Console.ReadLine();
}
private static void test4(object state)
{
Console.WriteLine("test4");
}
private static void test3(object state)
{
Console.WriteLine("test3");
}
private static void test2(object state)
{
while(true)
{
Console.WriteLine("test2");
Thread.Sleep(2000);
}
}
private static void test(object state)
{
while (true)
{
Console.WriteLine("test");
Thread.Sleep(2000);
wait.WaitOne();//只执行一次
}
}
}
}
本文参考链接：https://www.cnblogs.com/lexus/archive/2008/08/02/1258821.html
