已经一年多的时间没有使用多线程了.最近需要做一个异步发邮件的功能,为了给同事写一个例子,想起以前写的爬虫的时候,
一个带我的人给了我一个通用的模板类, 我觉得那个例子很好,于是我根据自己的理解,写出了如下的代码.
首先是一个通用的 计数器.包括 当前在运行的线程数Current,执行成功的数量Success,执行失败的数量Error,以及总完成数量Finish.
为了方便大家理解,我还是做一下简要的说明.大致过程是这样的: 首先有一批任务进来,需要把这些任务用多线程来处理.(我这里使用的是线程池.)
关于池的概念大家应该埠陌生.比如 数据库的连接池.
任务一个一个的分配,每个任务都是开启一个子线程单独执行的,一边分配任务,一边执行任务.在不停的分配任务的时候,会出现这样一种情况,
就是任务已经分配出去很多了,但是执行的不够快,这个时候需要主线程等待一下,过一会再分配新的任务.
我将上面的描述用一个日常生活中的例子来做说明: 一个小饭店只有10个座位,10点开始营业,到了12点吃饭的时间,外面来了100个客人,
挨个挨个的进入饭店就餐.
第一个人进,占位置1个,
第二个人进入, 占位置1个,
第三个人进入,占位置1个,
第四个人进入占位置1个,
第五个人进入占位置1个, 然后第一个人打包的走了,有个位置空出来了,
现在已经使用了 4个位置,至于进来以后坐什么地方,这个不用管.
第六个人进入占了位置1个,...第九个人进来了,占位置1个,第2个人吃完了,第十个人进来了,第是一个人进来了,第十二个人进来了,
第三个人吃完了,第十三个人进来了,
第十四个人想进来,抱歉,已经坐满了,要等待,过1分钟再看有没有位置,如果没有人离开,外面的人一直不能进来,
过了五分钟, 再看发现一个桌子四个人都离开了,于是第十四个人进来了,第十五个人进来了....后面的类似前面的...
废话啰嗦完了,上代码
1.先上一个计数器读写的类
public sealed class ThreadCounter
{
private int current = 0;
private int error = 0;
private int success = 0;
private int finish = 0;
private static object writeObj = new object();
public int getCurrent()
{
int res = 0;
lock (writeObj)
{
res = current;
}
return res;
}
public int getError()
{
int res = 0;
lock (writeObj)
{
res = error;
}
return res;
}
public int getSuccess()
{
int res = 0;
lock (writeObj)
{
res = success;
}
return res;
}
public int getFinish()
{
int res = 0;
lock (writeObj)
{
res = finish;
}
return res;
}
public void setAddCurrent()
{
lock (writeObj)
{
current++;
}
}
public void setSuccess()
{
lock (writeObj)
{
success++;
current--;
}
}
public void setError()
{
lock (writeObj)
{
error++;
current--;
}
}
public void setFinish()
{
lock (writeObj)
{
finish++;
}
}
}
2.主体运行业务处理类
public class RunManager
{
public RunManager() { }
public int ThreadCount { get; set; }
private static int MaxThreadCount = 10;
/// <summary>
/// 最大开启的线程数
/// </summary>
/// <param name="threadCount"></param>
public void RunStart(int threadMaxCount)
{
//最大开启的线程数
Console.WriteLine("设置最大线程数:" + threadMaxCount);
MaxThreadCount = threadMaxCount;
RunTask(100);
}
public void RunTask(int taskCount)
{
string mss = "";
string[] arr = new string[] {"|","/","-","\\" };
int j = 0;
while (true)
{
j++;
if (j==4)
{
j = 0;
}
Console.Clear();
Console.SetCursorPosition(1, 0);
Console.WriteLine("当前运行情况");
Console.Write("["+arr[j]+"]\t");
mss = string.Format("Query:{0} , Current:{1} , Success:{2} , Error:{3} , Finish:{4}", taskCount, tc.getCurrent(),tc.getSuccess(), tc.getError(), tc.getFinish());
Console.WriteLine(mss);
Thread.Sleep(500);
if (taskCount>0&&tc.getCurrent() < MaxThreadCount)
{
taskCount--;
tc.setAddCurrent();
ThreadPool.UnsafeQueueUserWorkItem(RuanCallBack, 0);
}
else if (taskCount==0)
{
taskCount = random.Next(60, 125);
}
else if (tc.getCurrent() >= MaxThreadCount)
{
Thread.Sleep(6000);
}
}
}
ThreadCounter tc = new ThreadCounter();
Random random = new Random();
public void RuanCallBack(object parms)
{
// 将传过来的参数, 整理为要查询的数据, 以及处理数据,写入文件,打包zip,
// 发送邮件通知用户下载
try
{
int t = random.Next(3, 7);
Thread.Sleep(t * 1000);
tc.setSuccess();
}
catch (Exception)
{
tc.setError();
}
finally
{
tc.setFinish();
}
//Console.WriteLine("当前线程序号:" + parms.ToString() + "\t线程停顿时间:" + t);
}
}
3.调用
static void Main(string[] args)
{
RunManager r = new RunManager();
r.RunStart(20);
Console.Read();
Console.WriteLine("运行结束");
}
先来看看运行效果
Nice!
原代码下载: 下载