C# 线程
目录
委托方式异步
启动方式
我们先声明一个方法：
private void doSomeThing(string name)
{
Console.WriteLine($"线程：{Thread.CurrentThread.ManagedThreadId} 开始执行doSomeThing({name}) ……");
Thread.Sleep(2000);
Console.WriteLine($"线程：{Thread.CurrentThread.ManagedThreadId} 结束执行doSomeThing({name}) ……");
}
再通过BeginInvoke()执行一个委托：
Console.WriteLine($"线程：{Thread.CurrentThread.ManagedThreadId} 开始执行Show()……");
Action<string> action = doSomeThing;
action.BeginInvoke("Oliver", null, null);
Console.WriteLine($"线程：{Thread.CurrentThread.ManagedThreadId} 结束执行Show()……");
输出如下：
线程：1 开始执行Show()……
线程：1 结束执行Show()……
线程：3 开始执行doSomeThing(Oliver) ……
线程：3 结束执行doSomeThing(Oliver) ……
判断异步线程执行完毕
我们通过上面的例子就很简单的执行了一个异步线程。但是我们如何判断异步线程执行完毕呢？
1、使用回掉函数
我们在使用BeginInvoke()方法时，它有三个参数，除了我们委托定义的一个参数外，它还有2个参数。
倒数第二个参数，就是异步调用执行完毕之后的回掉函数；最后一个参数是一个Object类型的，我们可以将这个参数传递到回调函数中。
回调函数它也有一个参数是IAsyncResult，这个参数经过我们比对和BeginInvoke()的返回值是一个对象。
通过下面的代码我们实现了在异步线程结束之后，我们调用了我们想调用的方法。
代码如下：
Console.WriteLine($"线程：{Thread.CurrentThread.ManagedThreadId} 开始执行Show()……");
Action<string> action = doSomeThing;
IAsyncResult asyncResult = null;
AsyncCallback callback = (ar) =>
{
Console.WriteLine(object.ReferenceEquals(ar, asyncResult));//结果为true，说明两个对象是同一个对象
Console.WriteLine($"ar.AsyncState：{ar.AsyncState}");
Console.WriteLine($"线程：{Thread.CurrentThread.ManagedThreadId}异步线程执行完毕了……");
};
asyncResult = action.BeginInvoke("Oliver", callback, "HI");
Console.WriteLine("……做一些其它的事儿……");
Console.WriteLine("……做一些其它的事儿……");
Console.WriteLine("……做一些其它的事儿……");
Console.WriteLine("……做一些其它的事儿……");
Console.WriteLine($"线程：{Thread.CurrentThread.ManagedThreadId} 结束执行Show()……");
输出如下：
线程：1 开始执行Show()……
……做一些其它的事儿……
……做一些其它的事儿……
……做一些其它的事儿……
……做一些其它的事儿……
线程：1 结束执行Show()……
线程：3 开始执行doSomeThing(Oliver) ……
线程：3 结束执行doSomeThing(Oliver) ……
ar.AsyncState：HI
线程：3异步线程执行完毕了……
2、使用asyncResult.IsCompleted
asyncResult.IsCompleted 表示异步操作是否已完成
Console.WriteLine($"线程：{Thread.CurrentThread.ManagedThreadId} 开始执行Show()……");
Action<string> action = doSomeThing;
IAsyncResult asyncResult = action.BeginInvoke("Oliver", null, null);
Console.WriteLine("……做一些其它的事儿……");
Console.WriteLine("……做一些其它的事儿……");
Console.WriteLine("……做一些其它的事儿……");
Console.WriteLine("……做一些其它的事儿……");
while (!asyncResult.IsCompleted)//判断线程是否执行完毕
{
Thread.Sleep(200);
}
//异步线程执行完毕才会执行下面代码，但是程序会处于假死状态
Console.WriteLine($"线程：{Thread.CurrentThread.ManagedThreadId} 结束执行Show()……");
输出：
线程：1 开始执行Show()……
……做一些其它的事儿……
……做一些其它的事儿……
……做一些其它的事儿……
……做一些其它的事儿……
线程：3 开始执行doSomeThing(Oliver) ……
线程：3 结束执行doSomeThing(Oliver) ……
线程：1 结束执行Show()……
3、使用EndInvoke()
当调用EndInvoke() 方法后，会阻塞当前线程。
Console.WriteLine($"线程：{Thread.CurrentThread.ManagedThreadId} 开始执行Show()……");
Action<string> action = doSomeThing;
IAsyncResult asyncResult = action.BeginInvoke("Oliver", null, null);
Console.WriteLine("……做一些其它的事儿……");
Console.WriteLine("……做一些其它的事儿……");
Console.WriteLine("……做一些其它的事儿……");
Console.WriteLine("……做一些其它的事儿……");
action.EndInvoke(asyncResult);
Console.WriteLine($"线程：{Thread.CurrentThread.ManagedThreadId} 结束执行Show()……");
输出：
线程：1 开始执行Show()……
……做一些其它的事儿……
……做一些其它的事儿……
……做一些其它的事儿……
……做一些其它的事儿……
线程：3 开始执行doSomeThing(Oliver) ……
线程：3 结束执行doSomeThing(Oliver) ……
线程：1 结束执行Show()……
4、使用WaitOne
使用asyncResult.AsyncWaitHandle.WaitOne() 等待当前线程执行完毕
Console.WriteLine($"线程：{Thread.CurrentThread.ManagedThreadId} 开始执行Show()……");
Action<string> action = doSomeThing;
IAsyncResult asyncResult = action.BeginInvoke("Oliver", null, null);
Console.WriteLine("……做一些其它的事儿……");
Console.WriteLine("……做一些其它的事儿……");
Console.WriteLine("……做一些其它的事儿……");
Console.WriteLine("……做一些其它的事儿……");
asyncResult.AsyncWaitHandle.WaitOne();//等待任务的完成
//asyncResult.AsyncWaitHandle.WaitOne(-1);//等待任务的完成
//asyncResult.AsyncWaitHandle.WaitOne(1000);//等待；但是最多等待1000ms
Console.WriteLine($"线程：{Thread.CurrentThread.ManagedThreadId} 结束执行Show()……");
输出：
线程：1 开始执行Show()……
……做一些其它的事儿……
……做一些其它的事儿……
……做一些其它的事儿……
……做一些其它的事儿……
线程：3 开始执行doSomeThing(Oliver) ……
线程：3 结束执行doSomeThing(Oliver) ……
线程：1 结束执行Show()……
Thread 线程
.netFrameWork 1.0 时出现的
使用方式
Thread 对象创建时，需要传递一个没有参数，没有返回值的委托进去。
Console.WriteLine("show 开始");
Thread thread = new Thread(() =>
{
Console.WriteLine("线程开始了....");
});
thread.Start();
Console.WriteLine("show 结束");
输出：
show 开始
show 结束
线程开始了....
其它
thread.Suspend();//线程挂起
thread.Resume();//唤醒线程
thread.Abort();//销毁，方式是抛异常 也不建议 不一定及时/有些动作发出收不回来
Thread.ResetAbort();//取消Abort异常
thread.Join();//当前线程等待thread完成
thread.Join(500);//最多等500
thread.IsBackground = true;//指定后台线程：随着进程退出。默认是前台线程，启动之后一定要完成任务的，阻止进程退出
thread.Priority = ThreadPriority.Highest;//线程优先级。CPU会优先执行 Highest 不代表说Highest就最先
ThreadPool 线程池
.netFrameWork 2.0 时出现的，它是对Thread的封装。
使用方式
public static bool QueueUserWorkItem(WaitCallback callBack);
public static bool QueueUserWorkItem(WaitCallback callBack, object state);
Task 任务
.netFrameWork 3.0 时出现的，是基于 ThreadPool的。
创建多线程的几种方式
- Task.Run(() => {……});
- Task.Factory.StartNew(() => {……});
- new Task(() => {……}).Start();
回调
单任务回调
//执行一个任务A后回调任务B
Task.Run(() => {
//任务A
}).ContinueWith(t => {
//任务B
});
多个任务执行完回调
List<Task> tlist = new List<Task>();
tlist.Add(Task.Run(() =>
{
Console.WriteLine("1-开始");
Thread.Sleep(1000);
Console.WriteLine("1-结束");
}));
tlist.Add(Task.Run(() =>
{
Console.WriteLine("2-开始");
Thread.Sleep(1000);
Console.WriteLine("2-结束");
}));
Task.Factory.ContinueWhenAll(tlist.ToArray(), (o) =>
{
Console.WriteLine("回调函数……");
});
主要API
| 对象 | 方法 | 说明 |
|---|---|---|
| Task | WaitAll | 等待全部任务完成，阻塞 |
| Task | WaitAny | 等待任意一个任务完成，阻塞 |
| Task | WhenAll | 所有任务完成返回一个Task，不阻塞 |
| Task | WhenAny | 任意一个任务完成返回一个Task，不阻塞 |
| Task | ContinueWith | 等待该任务完成，相当于回调 |
| Task | Delay | 延迟执行 |
| TaskFactory | ContinueWhenAll | 创建一个延续任务，该任务在一组指定的任务完成后开始。 |
| TaskFactory | ContinueWhenAny | 创建一个延续任务，它将在提供的组中的任何任务完成后马上开始。 |
Parallel 并行编程
基于 Task 实现的
卡界面
启动方式
- Parallel.Invoke(() => {}, () => {});
- Parallel.For(0, 5, i => {Console.WriteLine(i);});
- Parallel.ForEach(new string[] { "0","1","2","3","4"}, i => {Console.WriteLine(i);});
ParallelOptions
ParallelOptions 可以控制并发数量
ParallelOptions parallelOptions = new ParallelOptions();
parallelOptions.MaxDegreeOfParallelism = 3;//最大并发3个
其它
线程异常
比如我们有下面这么一段代码，里面写了try catch的，但是没有捕捉到异常信息，这是因为线程是异步执行的，当执行到catch下面的时候，线程中还没有抛出异常，所以这个就无法正常捕获异常。
正确的写法，应该是在方法体中进行 try catch。
try
{
Task.Run(() =>
{
throw new Exception("执行失败");
})
}catch (Exception ex)
{
Console.WriteLine(ex.Message);
}
Console.WriteLine("Hi");//当执行到这儿的时候可能，线程还没有执行。
线程取消
当我们在开发过程中可能遇到这种情况，同时执行50个任务，当其中任意一个任务执行失败的时候，我们就停止其它的任务执行。
我们通过下面的例子模拟这种情况，当任务10 执行失败后，会抛出异常。已经执行但未执行完毕的任务，当检测到信号量为取消的时候，可以通过程序的判断取消执行；尚未开始执行的任务系统会自动放弃执行，并抛出异常。
try
{
List<Task> taskList = new List<Task>();
//设置信号量
CancellationTokenSource cancellationTokenSource = new CancellationTokenSource();
for (int i = 0; i < 50; i++)
{
int k = i;
taskList.Add(Task.Factory.StartNew((t) =>
{
try
{
if (t.ToString() == "10")//模拟任务10 执行失败
{
throw new Exception(string.Format("{0} 执行失败", t));
}
Thread.Sleep(200);//模拟程序操作时间
if (cancellationTokenSource.IsCancellationRequested)//判断信号量是否为取消
{
//如果信号量为取消就不执行了
Console.WriteLine(string.Format("{0} 取消执行", t));
}
else
{
Console.WriteLine(string.Format("{0} 执行成功", t));
}
}
catch (Exception ex)
{
//当执行任务异常时，将信号量设置为取消
cancellationTokenSource.Cancel();
Console.WriteLine(ex.Message);
}
}, k, cancellationTokenSource.Token));//将信号量传递给创建任务的操作。
}
Task.WaitAll(taskList.ToArray());
}
catch (AggregateException aex)//当系统检测到信号量为取消时单是队列中还有未执行的任务，会抛出这个异常
{
//因为是并行多线程的操作，所以会抛出一个异常列表。
foreach (var item in aex.InnerExceptions)
{
Console.WriteLine(item.Message);
}
}
临时变量
在使用多线程时有时会产生莫名奇妙的异常，很有可能就是因为临时变量的问题。
错误实例
for (int i = 0; i < 5; i++)
{
Task.Factory.StartNew(() =>
{
Console.WriteLine(i); //输出 5 5 5 5 5。输出的信息与想象的不一样，是因为当前上下文中只有一个 i，i是不断变化的
});
}
正确实例
for (int i = 0; i < 5; i++)
{
int k = i;//每一次循环都会重新定义一个k
Task.Factory.StartNew(() =>
{
Console.WriteLine(k); //输出 0 2 1 4 3。输出的信息与想象的一致，因为当前上下文的每一个线程都有一个k，相当于每一个任务都有一个独立的k。
});
}
线程安全
当我们多个线程同时访问一个变量的时候，可能因为操作不同步造成线程间的不安全。例如以下代码：
List<Task> taskList = new List<Task>();
int totalCount = 0;
for (int i = 0; i < 10000; i++)
{
taskList.Add(Task.Factory.StartNew(() =>
{
totalCount += 1;
}));
}
Task.WaitAll(taskList.ToArray());
Console.WriteLine(totalCount);//输出9985
Lock
优点：因为只有一个线程可以进去，解决问题
缺点：没有并发，牺牲了性能
代码
先定义一个锁对象
private static readonly object thread_Lock = new object();
然后再调用共享对象的时候使用锁包起来，就没有问题了。
taskList = new List<Task>();
totalCount = 0;
for (int i = 0; i < 10000; i++)
{
taskList.Add(Task.Factory.StartNew(() =>
{
lock (thread_Lock)
{
totalCount += 1;
}
}));
}
Task.WaitAll(taskList.ToArray());
Console.WriteLine(totalCount);//输出10000
不要冲突
数据拆分，避免冲突。
如我们需要并发的时候写日志的时候，当我们将所有的日志写到一个文件的时候就会发生线程的不安全问题。但是如果使用Lock的时候又会将并发变为单线程，牺牲了性能，这种情况我们就可以将日志写到多个文件，等所有线程执行完毕之后再将多个文件合并为一个文件。
WinForm中线程问题:线程间操作无效: 从不是创建控件
- 可以在load时将CheckForIllegalCrossThreadCalls 属性的值设置为 false。这样进行非安全线程访问时，运行环境就不去检验它是否是线程安全的。
- 交给UI处理操作
this.Invoke(new Action(() =>
{
lbl.Text = text;
}));//交给UI线程去更新
应用
写线程的文章感觉也不知从何写起，里面大多数是枯燥的API，我在这里给大家举个小小的应用。
这个应用分别使用 Thread、ThreadPool 、Task、Parallel 来实现，我们或许可以从中比较出他们的差异。
需求
我之前做过一个小说网站：http://www.twgdh.com 这个里面的主要数据就是使用多线程抓取的。
比如我们现在要在网上抓取一本小说的内容，然后存放起来。
公共代码
HttpHelper
/// <summary>
/// 主要处理请求相关的东西
/// </summary>
public class HttpHelper
{
public static string GetUrltoHtml(string Url, string type)
{
System.Net.WebRequest wReq = System.Net.WebRequest.Create(Url);
// Get the response instance.
System.Net.WebResponse wResp = wReq.GetResponse();
System.IO.Stream respStream = wResp.GetResponseStream();
// Dim reader As StreamReader = New StreamReader(respStream)
using (System.IO.StreamReader reader = new System.IO.StreamReader(respStream, Encoding.GetEncoding(type)))
{
return reader.ReadToEnd();
}
}
}
FileHelper
/// <summary>
/// 主要处理文件相关的操作
/// </summary>
public class FileHelper
{
public void CreateFile(string path, string content)
{
using (FileStream fileStream = File.Create(path))//打开文件流 （创建文件并写入）
{
StreamWriter sw = new StreamWriter(fileStream);
sw.WriteLine(content);
sw.Flush();
}
}
}
QiDianHelper
/// <summary>
/// 抓取小说网站的数据
/// </summary>
public class QiDianHelper
{
/// <summary>
/// 根据书号得到本书的所有章节信息
/// </summary>
/// <param name="bid"></param>
/// <returns></returns>
public List<Volume> GetVolume(string bid)
{
string url = "https://read.qidian.com/ajax/book/category?bookId=" + bid;
string json = HttpHelper.GetUrltoHtml(url, "utf-8");
qidian_json qidian_Data = Newtonsoft.Json.JsonConvert.DeserializeObject<qidian_json>(json);
List<Volume> volumeList = qidian_Data.data.vs;
return volumeList;
}
/// <summary>
/// 更新章节
/// </summary>
/// <param name="chapter"></param>
/// <returns></returns>
public void UpdateChapter(Chapter chapter,string bookNo)
{
string url = "https://vipreader.qidian.com/chapter/"+ bookNo + "/"+ chapter.id;
string html = HttpHelper.GetUrltoHtml(url, "utf-8");
string pattern = @"<div class=""read-content j_readContent"">(?<content>[sS]*?)</div>";
RegexOptions options = RegexOptions.Multiline;
foreach (Match m in Regex.Matches(html, pattern, options))
{
chapter.Content = m.Groups["content"].Value;
chapter.status = "更新成功";
return;
}
chapter.status = "更新失败";
}
}
Model
/// <summary>
/// 主要用于反序列化
/// </summary>
public class qidian_json
{
public qidian_data data { get; set; }
public int code { get; set; }
}
/// <summary>
/// 主要用于反序列化
/// </summary>
public class qidian_data
{
public List<Volume> vs { get; set; }
}
/// <summary>
/// 卷
/// </summary>
public class Volume
{
public int cCnt { get; set; }
public bool hS { get; set; }
public int isD { get; set; }
public int vId { get; set; }
public string vN { get; set; }
public int vS { get; set; }
public int wC { get; set; }
public List<Chapter> cs { get; set; }
public int Index { get; set; }
}
/// <summary>
/// 章节
/// </summary>
public class Chapter
{
public int Index { get; set; }
public string cN { get; set; }
public string cU { get; set; }
public int cnt { get; set; }
public int id { get; set; }
public int sS { get; set; }
public DateTime uT { get; set; }
public int uuid { get; set; }
/// <summary>
/// 存放章节内容信息
/// </summary>
public string Content { get; set; }
public string status { get; set; }
}
Thread 实现
代码
public class ThreadTest
{
//下载完毕后创建日志时使用
List<string> contentList = new List<string>();
public void RequestBook(string bookNo)
{
Stopwatch watch = new Stopwatch();
watch.Start();
this.output("开始更新书籍。书号：" + bookNo);
QiDianHelper qiDianHelper = new QiDianHelper();
this.output("开始查询卷信息……");
var vList = qiDianHelper.GetVolume(bookNo);
this.output("卷信息查询完毕，共查询到{0}卷，{1}章……", vList.Count, vList.Sum(c => c.cCnt));
this.output("开始查询章节信息……");
FileHelper fileHelper = new FileHelper();
string path = Path.Combine(AppDomain.CurrentDomain.BaseDirectory, "Thread");
if (!System.IO.Directory.Exists(path))
{
System.IO.Directory.CreateDirectory(path);
}
//我们启动多线程进行查询,开启30个线程
List<Thread> tList = new List<Thread>();
int currentThreadCount = 0;
int vIndex = 0;
int cIndex = 0;
foreach (var v in vList)//循环卷
{
v.Index = ++vIndex;
cIndex = 0;
foreach (var c in v.cs)//循环卷中的章节
{
c.Index = ++cIndex;
while (currentThreadCount >= 30)//当同时使用的线程超过30个后，我们将主线程停止50毫秒再继续判断
{
Thread.Sleep(50);
}
//创建一个线程
Thread thread = new Thread(() =>
{
try
{
//this.output("开始更新 {0} {1} ", v.vN, c.cN);
qiDianHelper.UpdateChapter(c, bookNo);
this.output("{0} {1} {2}", v.vN, c.cN, c.status);
//写入文件
fileHelper.CreateFile(Path.Combine(path, string.Format("{2}_{3} {0}{1}.txt", v.vN, c.cN, v.Index.ToString("00"), c.Index.ToString("000"))), c.Content);//将下载的结果存储txt中
currentThreadCount--;//线程执行完毕后更新变量
}
catch (Exception ex)
{
this.output("{0} {1} {2}", v.vN, c.cN, "更新异常！");
currentThreadCount--;//线程异常时更新这个变量
}
});
thread.Start();//开始这个线程
tList.Add(thread);//将线程存入线程列表，方便确定线程是否执行完毕
currentThreadCount++;//线程开始之后，更新这个变量
}
}
foreach (var item in tList)
{
item.Join();//这儿会阻塞当前线程，因为循环的所有线程 所以这个循环走完，所有线程都会执行完毕。
}
this.output("书籍更新完毕。");
watch.Stop();
this.output("用时：{0}秒", watch.ElapsedMilliseconds / 1000);
fileHelper.CreateFile(Path.Combine(path, "log.txt"), string.Join("
", contentList.ToArray()));
}
/// <summary>
/// 输出信息
/// </summary>
/// <param name="format"></param>
/// <param name="arg"></param>
private void output(string format, params object[] arg)
{
string content = string.Format("{0} 线程ID:【{2}】 {1}", DateTime.Now.ToLocalTime(), string.Format(format, arg), Thread.CurrentThread.ManagedThreadId);
contentList.Add(content);
Console.WriteLine(content);
}
}
调用
调用时DefaultConnectionLimit：可以参照多线程环境下调用 HttpWebRequest 并发连接限制
//获取或设置 System.Net.ServicePoint 对象所允许的最大并发连接数。
System.Net.ServicePointManager.DefaultConnectionLimit = 100;
string bookNo = "1010468795";//这个是某一本书的书号
ThreadTest threadTest = new ThreadTest();
threadTest.RequestBook(bookNo);
ThreadPool 实现
代码
/// <summary>
/// 线程池
/// </summary>
public class ThreadPoolTest
{
//下载完毕后创建日志时使用
List<string> contentList = new List<string>();
public void RequestBook(string bookNo)
{
Stopwatch watch = new Stopwatch();
watch.Start();
this.output("开始更新书籍。书号：" + bookNo);
QiDianHelper qiDianHelper = new QiDianHelper();
this.output("开始查询卷信息……");
var vList = qiDianHelper.GetVolume(bookNo);
this.output("卷信息查询完毕，共查询到{0}卷，{1}章……", vList.Count, vList.Sum(c => c.cCnt));
this.output("开始查询章节信息……");
FileHelper fileHelper = new FileHelper();
string path = Path.Combine(AppDomain.CurrentDomain.BaseDirectory, "ThreadPool");
if (!System.IO.Directory.Exists(path))
{
System.IO.Directory.CreateDirectory(path);
}
//我们启动多线程进行查询,设置程序最多开启30个线程
ThreadPool.SetMaxThreads(30, 30);
//信号量列表，用于阻塞主线程
List<ManualResetEvent> manualEvents = new List<ManualResetEvent>();
int vIndex = 0;
int cIndex = 0;
foreach (var v in vList)
{
v.Index = ++vIndex;
cIndex = 0;
foreach (var c in v.cs)
{
ManualResetEvent manualResetEvent = new ManualResetEvent(false);//定义信号量
manualEvents.Add(manualResetEvent);//将定义的信号量添加到列表
c.Index = ++cIndex;
ThreadPool.QueueUserWorkItem(o =>
{
try
{
this.output("开始更新 {0} {1} ", v.vN, c.cN);
qiDianHelper.UpdateChapter(c, bookNo);
this.output("{0} {1} {2}", v.vN, c.cN, c.status);
//写入文件
fileHelper.CreateFile(Path.Combine(path, string.Format("{2}_{3} {0}{1}.txt", v.vN, c.cN, v.Index.ToString("00"), c.Index.ToString("000"))), c.Content);//将下载的结果存储txt中
}
catch (Exception ex)
{
this.output("{0} {1} {2}", v.vN, c.cN, "更新异常！");
}
finally
{
manualResetEvent.Set();//表明该线程已经结束
}
});
}
}
foreach (var item in manualEvents)//等待线程池中的全部线程结束
{
item.WaitOne();
}
this.output("书籍更新完毕。");
watch.Stop();
this.output("用时：{0}秒", watch.ElapsedMilliseconds / 1000);
//输出日志
fileHelper.CreateFile(Path.Combine(path, "log.txt"), string.Join("
", contentList.ToArray()));
}
/// <summary>
/// 输出信息
/// </summary>
/// <param name="format"></param>
/// <param name="arg"></param>
private void output(string format, params object[] arg)
{
string content = string.Format("{0} 线程ID:【{2}】 {1}", DateTime.Now.ToLocalTime(), string.Format(format, arg), Thread.CurrentThread.ManagedThreadId);
contentList.Add(content);
Console.WriteLine(content);
}
}
Task 实现
代码
//下载完毕后创建日志时使用
List<string> contentList = new List<string>();
public void RequestBook(string bookNo)
{
Stopwatch watch = new Stopwatch();
watch.Start();
this.output("开始更新书籍。书号：" + bookNo);
QiDianHelper qiDianHelper = new QiDianHelper();
this.output("开始查询卷信息……");
var vList = qiDianHelper.GetVolume(bookNo);
this.output("卷信息查询完毕，共查询到{0}卷，{1}章……", vList.Count, vList.Sum(c => c.cCnt));
this.output("开始查询章节信息……");
FileHelper fileHelper = new FileHelper();
string path = Path.Combine(AppDomain.CurrentDomain.BaseDirectory, "Task");
if (!System.IO.Directory.Exists(path))
{
System.IO.Directory.CreateDirectory(path);
}
ThreadPool.SetMaxThreads(30, 30);//因为Task是基于ThreadPool 的，所以设置这个对Task也有效
List<Task> taskList = new List<Task>();//任务列表，阻塞主线程时使用
int vIndex = 0;
int cIndex = 0;
foreach (var v in vList)
{
v.Index = ++vIndex;
cIndex = 0;
foreach (var c in v.cs)
{
c.Index = ++cIndex;
var task = Task.Factory.StartNew(() =>
{
try
{
this.output("开始更新 {0} {1} ", v.vN, c.cN);
qiDianHelper.UpdateChapter(c, bookNo);
this.output("{0} {1} {2}", v.vN, c.cN, c.status);
//写入文件
fileHelper.CreateFile(Path.Combine(path, string.Format("{2}_{3} {0}{1}.txt", v.vN, c.cN, v.Index.ToString("00"), c.Index.ToString("000"))), c.Content);//将下载的结果存储txt中
}
catch (Exception ex)
{
this.output("{0} {1} {2}", v.vN, c.cN, "更新异常！");
}
});
taskList.Add(task);//添加到任务列表中
}
}
Task.WaitAll(taskList.ToArray());//等待所有任务执行完毕
this.output("书籍更新完毕。");
watch.Stop();
this.output("用时：{0}秒", watch.ElapsedMilliseconds / 1000);
//输出日志
fileHelper.CreateFile(Path.Combine(path, "log.txt"), string.Join("
", contentList.ToArray()));
}
/// <summary>
/// 输出信息
/// </summary>
/// <param name="format"></param>
/// <param name="arg"></param>
private void output(string format, params object[] arg)
{
string content = string.Format("{0} 线程ID:【{2}】 {1}", DateTime.Now.ToLocalTime(), string.Format(format, arg), Thread.CurrentThread.ManagedThreadId);
contentList.Add(content);
Console.WriteLine(content);
}
Parallel 实现
代码
public class ParallelTest
{
//下载完毕后创建日志时使用
List<string> contentList = new List<string>();
public void RequestBook(string bookNo)
{
Stopwatch watch = new Stopwatch();
watch.Start();
this.output("开始更新书籍。书号：" + bookNo);
QiDianHelper qiDianHelper = new QiDianHelper();
this.output("开始查询卷信息……");
var vList = qiDianHelper.GetVolume(bookNo);
this.output("卷信息查询完毕，共查询到{0}卷，{1}章……", vList.Count, vList.Sum(c => c.cCnt));
this.output("开始查询章节信息……");
FileHelper fileHelper = new FileHelper();
string path = Path.Combine(AppDomain.CurrentDomain.BaseDirectory, "Parallel");
if (!System.IO.Directory.Exists(path))
{
System.IO.Directory.CreateDirectory(path);
}
ParallelOptions parallelOptions = new ParallelOptions();
parallelOptions.MaxDegreeOfParallelism = 30;//控制并发数为30；
List<Action> actionList = new List<Action>();//用于保存并发执行的操作
int vIndex = 0;
int cIndex = 0;
foreach (var v in vList)
{
v.Index = ++vIndex;
cIndex = 0;
foreach (var c in v.cs)
{
c.Index = ++cIndex;
Action action = () =>
{
try
{
this.output("开始更新 {0} {1} ", v.vN, c.cN);
qiDianHelper.UpdateChapter(c, bookNo);
this.output("{0} {1} {2}", v.vN, c.cN, c.status);
//写入文件
fileHelper.CreateFile(Path.Combine(path, string.Format("{2}_{3} {0}{1}.txt", v.vN, c.cN, v.Index.ToString("00"), c.Index.ToString("000"))), c.Content);//将下载的结果存储txt中
}
catch (Exception ex)
{
this.output("{0} {1} {2}", v.vN, c.cN, "更新异常！");
}
};
actionList.Add(action);//将操作添加到列表中
}
}
Parallel.Invoke(actionList.ToArray());//Parallel会开启多线程执行actionList 列表中的方法，执行的时候默认会阻塞主线程，所以不用人工阻塞。
this.output("书籍更新完毕。");
watch.Stop();
this.output("用时：{0}秒", watch.ElapsedMilliseconds / 1000);
//输出日志
fileHelper.CreateFile(Path.Combine(path, "log.txt"), string.Join("
", contentList.ToArray()));
}
/// <summary>
/// 输出信息
/// </summary>
/// <param name="format"></param>
/// <param name="arg"></param>
private void output(string format, params object[] arg)
{
string content = string.Format("{0} 线程ID:【{2}】 {1}", DateTime.Now.ToLocalTime(), string.Format(format, arg), Thread.CurrentThread.ManagedThreadId);
contentList.Add(content);
Console.WriteLine(content);
}
}