.NET异步程序设计之任务并行库
目錄
- 1.簡介
- 2.Parallel類
- 2.0 Parallel類簡介
- 2.1 Parallel.For()
- 2.2 Parallel.ForEach()
- 2.3 Parallel.Invoke()
- 2.4 補充:線程安全集合
- 3.Task類
- 3.0 Task類簡介
- 3.1 創建無返回值的Task任務
- 3.2 創建有返回值的Task任務
- 3.3 為Task添加延續任務
- 3.4 Task.Delay
- 3.5 Task對象的其他一些靜態方法
- 3.6 取消異步操作
- 4.并行Linq(PLinq)
- 4.1 AsParallel()
- 4.2 取消并行查詢
- 5.參考&源代碼下載
shanzm-2020年2月16日 00:45:04
1.簡介
System.Threading.Tasks中的類型被稱為任務并行庫(Task Parallel Library,TPL)。
System.Thread.Tasks?命名空間是.NET Framework4.0所提供,
“TPL使用CLR線程池(TPL是TreadPool的封裝),自動將應用程序的工作,動態分配到可用的CPU中。TPL還處理工作分區、線程調度、狀態管理和其他低級別的細節操作。最終結果是,你可以最大限度地提升.NET應用程序的性能,并且避免直接操作線程所帶來的復雜性” --《精通C#》
?
2.Parallel類
?
2.0 Parallel類簡介
在System.Threading.Tasks命名空間下有一個靜態類:Parallel類
Parallel可以實現對實現了IEnumerable接口的數據集合的每一個元素并行操作
有一點要說明的:并行操作會帶來一定的成本,如果任務本身能很快完成,或是循環次數很少,那么并行處理的速度也許會比非并行處理還慢。
Parallel類就只有三個方法:Parallel.For()、Parallel.ForEach()和Parallel.Invoke()
但是呢,這每個方法都有大量的重載(F12-->自行查看Parallel定義)
?
2.1 Parallel.For()
使用Parallel.For()可以對數組中的每一個元素進行并行操作
正常的遍歷數組是按照索引的順序執行的;但是在并行操作下,對數組的每一個元素的操作不一定按照索引順序操作
Parallel.For(),第一個參數是循環開始的索引(包含),第二個參數是循環結束的索引(不含)
Parallel.For()的第三個參數是一個有參數無返回值的委托,其參數是數組的索引
其實就相當于:for (int i = 0; i < length; i++)的異步版本,只是在這里是并行操作,所以并不按照數組中元素的順序執行,具體的執行順序是不可控的。
示例
static void Main(string[] args) {int[] intArray = new int[] { 1, 2, 3, 4, 5, 6, 7, 8, 9, 10 };Console.WriteLine("------------常規,對數組進行循環遍歷------------");Array.ForEach(intArray, n => Console.WriteLine($"當前操作的數組元素是{n}"));//注意這里的參數n是元素而不是索引Console.WriteLine("------------并行操作 對數組進行循環遍歷------------");Parallel.For(0, intArray.Length, (i) => Console.WriteLine($"當前循環次數{i},當前操作的數組元素是{intArray[i]}"));Console.ReadKey(); }運行結果:可以看出,對數組的元素的操作順序并不是按照索引的順序,而是不確定的。
?
2.2 Parallel.ForEach()
Parallel.ForEach()用于對泛型可枚舉對象的元素進行并行操作
其實就相當于:foreach (var item in collection)的異步版本
Parallel.ForEach()有大量的重載,這里展示一個簡單的操作
Parallel.ForEach()的第一個參數是待操作的可枚舉對象,第二個參數是一個有參數無返回值的委托,該委托參數是集合的元素(而不是索引)
示例
List<int> intList = new List<int>() { 1, 2, 3, 4, 5, 6, 7, 8, 9, 10 }; Parallel.ForEach(intList, n => Console.WriteLine(n+100)); Console.ReadKey();?
2.3 Parallel.Invoke()
Parallel.Invoke()對指定一系列操作并行運算
參數是一個Action委托數組(注意只能是Action[],即只能是無返回值的委托數組)
Parallel.Invoke()最常見用于并發請求接口
示例:
static void Main(string[] args) {Action action1=() =>{for (int i = 0; i < 5; i++){Console.WriteLine($"action-1-操作");}};Action action2 = () =>{for (int i = 0; i < 5; i++){Console.WriteLine($"action-2-操作");}};//Parallel.Invoke(action1, action2);Action[] actions = { action1, action2 };Parallel.Invoke(actions);Console.ReadKey(); }運行結果:
?
2.4 補充:線程安全集合
詳細可以參考微軟的在線文檔
多線程對同一個數據集合同時讀寫操作,可能會造成數據的混亂
.NET4 引入了System.Collections.Concurrent?命名空間,其中包含多個線程安全的數據集合類型。
現在的新項目中,只要是對數據集合進行多線程的增刪操作,就應該使用并發集合類。
但是,如果僅從集合進行多線程的讀取,則可使用一般的數據集合,即?System.Collections.Generic?命名空間中的類。
.net 中線程安全的數據集合有一下一些:
| BlockingCollection | 為實現?IProducerConsumerCollection?的所有類型提供限制和阻止功能。 有關詳細信息,請參閱?BlockingCollection 概述。 |
| ConcurrentDictionary | 鍵值對字典的線程安全實現。 |
| ConcurrentQueue | FIFO(先進先出)隊列的線程安全實現。 |
| ConcurrentStack | LIFO(后進先出)堆棧的線程安全實現。 |
| ConcurrentBag | 無序元素集合的線程安全實現。 |
| IProducerConsumerCollection | 類型必須實現以在?BlockingCollection?中使用的接口。 |
一個簡單的示例:給一個數據集合添加大批量的數據
List<int> list = new List<int>(); Parallel.For(0, 1000000, t => list.Add(t));若是按照上面使用Parallel.For()的并行方式給List添加數據,
則會報錯:“索引超出了數組界限。”或“ 源數組長度不足。請檢查 srcIndex 和長度以及數組的下限。”
即使沒有報錯,list中的數據也是有問題的(比可能數量不足)
當然可以通過加鎖的方式進行彌補:
List<int> list = new List<int>(); object locker = new object(); Parallel.For(0, 1000000, t => { lock(locker) { list.Add(t); } });這樣通過對操作的線程枷鎖,完全是沒有必要的,你可以使用線程安全的集合類型,比如在這里使用ConcurrentBag
ConcurrentBag<int> cBag = new ConcurrentBag<int>(); Parallel.For(0, 100000, t => cBag.Add(t));當然因為是并行操作,所以插入集合中的數據并不是按照0-100000的順序(僅僅是成段的有序)。
?
?
3.Task類
?
3.0 Task類簡介
System.Threading.Tasks命名空間中Task類,表示異步操作。
Task類可以輕松地在次線程中調用方法,可以作為異步委托的簡單替代品。
同時在該命名空間還有一個泛型Task<TResul>類,TResult 表示異步操作執行完成后返回值的類型。
創建一個Task操作,只需要使用靜態函數Task.Run()即可,
Task.Run()是一個.net framework4.5及以上定義的一個默認異步操作,
Task.Run()參數是委托,即需要異步執行的方法,
注意作為Task.Run()的參數的委托都是無參委托,
若Task.Run()參數是無返回值的委托Action,則Task.Run()返回值是Task類型
若Task.Run()參數是有返回值的委托Func<TResult>,則Task.Run()返回值是Task<TResult>泛型
注意:若是低于.net4.5,則可以使用Task.Factory.StartNew(),和Task.Run()靜態方法作用一樣
總而言之,言而總之,show you code ,一切皆明了!
?
3.1 創建無返回值的Task任務
示例:無返回值的Task
static void Main(string[] args) {//1.使用Task構造函數創建,必須顯式的使用.Start()才能開始執行//Task task = new Task(() => { Thread.Sleep(10); Console.WriteLine("我是Task ,我結束了"); });//task.Start();//2.使用TaskFactory.StartNew(工廠創建) 方法//Task task = Task.Factory.StartNew(() => { Thread.Sleep(10); Console.WriteLine("我是Task ,我結束了"); });//3.使用Task.Run()Task task = Task.Run(() => { Thread.Sleep(10); Console.WriteLine("我是Task.Run ,我結束了"); });if (!task.IsCompleted)//task.IsCompleted判斷當前的任務是否已完成{Console.WriteLine("當前的Task.Run()尚未執行完,但是因為異步,返回到調用函數,所以可以先執行后續的代碼");}Console.WriteLine("當前Task.Run還沒有完成,我們是在他之后的代碼但是先執行了");task.Wait();//強行鎖定線程,等待task完成Console.WriteLine("終于Task.Run完成了工作");Console.ReadKey(); }?
3.2 創建有返回值的Task任務
若是Task任務有返回值,返回值類型為Task<T>,使用返回值的Result屬性查詢具體值
調試時注意查看,運行到?Console.WriteLine(task.Result)的時候,其中Task任務還是在執行Thread.Sleep(1000)
還沒有出結果,我們希望的異步執行也沒有發生,而是程序是在一直在等待,這是為什么呢?
是因為一但執行了task.Result,即使task任務還沒有完成,主線程則停下等待,直到等待task.Result出結果
這種情況和異步委托中調用EndInvoke()是一樣的:一旦運行EndInvoke,若是引用方法還沒有完成,主線程則停止,直到引用函數運行結束。
所以可以這樣理解:task.Result可以看作是一個未來結果(一定有結果但還在運算中)
示例:有返回值的Task
static void Main(string[] args) {Console.WriteLine("SomeDoBeforeTask");Func<int> Do = () => { Thread.Sleep(1000); Console.WriteLine("Task.Run結束"); return 2; };Task<int> task = Task.Run(Do);Console.WriteLine(task.Status);//使用task.Status查看當前的Task的狀態:當前的狀態:WaitingToRunConsole.WriteLine(task.Result);//使用task.result操作Task任務的返回值:返回值是:2Console.WriteLine(task.Status);//使用task.Status查看當前的Task的狀態:當前的狀態:RanToComplationConsole.WriteLine("SomeDoAfterTask");Console.ReadKey(); }運行結果:
說明:
其中我們使用task.Result查看當前的task的狀態,其中Task的狀態(即其生命周期)只有三種:
- Created(創建Task):注意只有Task task=new Task(...),此時的Task狀態為Created,其他方式創建的Task跳過了Created狀態
- WaitingToRun(等待執行Task)
- RanToComplation(Task完成)
?
3.3 為Task添加延續任務
Task任務是在后臺執行的同時,主線程的繼續執行后續程序
所以有時候需要在Task結束后,繼續執行某個特定的任務,即為Task添加延續任務(也稱接續工作)
舉一個簡單的例子,
求解1-5000能求被3整除的個數,這個過程需要許多時間,我把它定義為一個Task.Run()
我們需要在求出結果后打印出結果,這里怎么操作呢?
若是直接使用task.Result則會阻塞主線程,一直等待運算出結果,這顯然不是我們想要的
若是使用while(!task.IsComplation){//后續操作},你無法判斷Task何時結束,而且一旦Task結束則會中斷后續操作
這里就是需要為Task加上接續工作
這里你可以明白,接續本質和異步委托中的回調模式是一樣的,回調方法就是接續工作
?
3.3.1使用task.ContinueWith()
task1.ContinueWith(...task2..)表示當task1結束后接著運行task2任務
注意這里我們使用Lambda表達式編寫接續工作,接續工作是有一個參數的,參數是Task類型,即上一個Task
即第一個Task完成后自動啟動下一個Task,實現Task的延續
注意:ContinueWith()的返回值亦是Task類型對象,即新創建的任務
可以為接續工作task2繼續添加接續工作task3
示例5 :
static void Main(string[] args){Console.WriteLine("task執行前...");Task<int> task1 = Task.Run(() => Enumerable.Range(1, 5000).Count(n => (n % 3) == 0));Task task2 = task1.ContinueWith(t => Console.WriteLine($"當你看到這句話則task1結束了,1-5000中能被3整除的個數{t.Result}"));//這里的t就是task1Task task3 = task2.ContinueWith(t => Console.WriteLine($"當你看到這句話則task2也結束了"));Console.WriteLine($"task1及其接續工作正在執行中," + "\t\n" + "我們現在正在執行其他的后續代碼");Console.ReadKey();}運行結果:
?
3.3.2使用Awaiter
使用task.GetAwaiter()為相關的task創建一個等待者
示例:
static void Main(string[] args){Console.WriteLine("task執行前...");Task<int> task1 = Task.Run(() => Enumerable.Range(1, 5000).Count(n => (n % 3) == 0));var awaiter = task1.GetAwaiter();//創建一個awaiter對象//awaiter.OnCompleted(() => Console.WriteLine($"當你看到這句話則task1結束了,1-5000中能被3整除的個{task1.Result}"));awaiter.OnCompleted(() => Console.WriteLine($"當你看到這句話則task1結束了,1-5000中能被3整除的個{awaiter.GetResult()}"));Console.WriteLine($"task1及其接續工作正在執行中," + "\t\n" + "我們現在正在執行其他的后續代碼");Console.ReadKey();}運行效果同上。
3.3.3使用ContinueWith和Awaiter的區別:
ContinueWith會返回Task對象,它非常適合用于增加更多的接續工作,不過,如果Task出錯,必須直接處理AggregateException。
使用task.GetAwaiter創建awaiter對象,是在.net4.5之后,其中C#5.0的異步功能就是使用這種方式。
使用awaiter也是可以使用task.Result直接的查看任務的結果,但是使用awaiter.GetResult()可以在Task出現異常的時候直接拋出,不會封裝在AggregateException中。
3.4 Task.Delay
延時執行Task
3.4.1 使用Task.Delay()和ContinueWith實現延遲工作
其實就相當于實現Thread.Sleep()的異步版本
若是你使用Thread.Sleep(),則會程序一直在等待(即阻塞線程),直到等待結束才會運行后續的代碼
而這里就相當于給給Thread.Sleep()一個加了接續工作,且這個接續工作是異步的。
即使用Task.Delay()不會阻塞主線程,主線程可以繼續執行后續代碼
示例:
//新建異步任務,30毫秒秒后執行Task.Delay(30).ContinueWith(c =>{for (int i = 0; i < 50; i++){Console.WriteLine(i + "這是Task在運行");}});for (int i = 0; i < 100; i++){Console.WriteLine(i + "這是Task之后的程序在運行");}調試的時候你會發現,剛開始的時候的時候是先顯示的"i這是Task之后的程序在運行"
之后在等帶了30毫秒,后就會開始顯示"i這是Task在運行"和"i這是Task之后的程序在運行"交叉顯示
運行結果如下:
?
3.4.2 使用Task.Delay()和Awaiter實現延遲工作
示例:運行效果同上
Task.Delay(30).GetAwaiter().OnCompleted(() =>{for (int i = 0; i < 50; i++){Console.WriteLine(i + "這是Awaiter在運行行");}});for (int i = 0; i < 100; i++){Console.WriteLine(i + "這是Awaiter之后的程序在運行行");}Console.ReadKey();?
3.5 Task對象的其他一些靜態方法
| Task.Wait | task1.Wait();就是等待任務執行(task1)完成,task1的狀態變為Completed |
| Task.WaitAll | 待所有的任務都執行完成 |
| Task.WaitAny | 發同Task.WaitAll,就是等待任何一個任務完成就繼續向下執行 |
| CancellationTokenSource | 通過cancellation的tokens來取消一個Task |
?
3.6 取消異步操作
異步方法是可以請求終止運行的,
System.Threading.Tasks命名空間中有兩個類是為此目的而設計的:Cance1lationToken和CancellationTokenSource。
下面看使用CancellationTokenSource和CancellationToken來實現取消某個異步操作。
這里使用Task.Run()為例,其第一個參數是一個Action委托,第二個參數就是CancellationToken對象
static void Main(string[] args) {CancellationTokenSource cts = new CancellationTokenSource();//生成一個CancellationTokenSource對象,該對象可以創建CancellationToken CancellationToken ct = cts.Token;//獲取一個令牌(token)Task.Run(() =>{for (int i = 0; i < 20; i++){if (ct.IsCancellationRequested){return;}Thread.Sleep(1000);Console.WriteLine($"異步程序的的循環:{i}");}}, ct);//注意Run()的第二個參數就是終止令牌tokenfor (int i = 0; i < 4; i++){Thread.Sleep(1000);Console.WriteLine($"主線程中循環:{i}");}Console.WriteLine("馬上sts.Cancel(),即將要終止異步程序");cts.Cancel();//含有該CancellationTokenSource的token的異步程序,終止!Console.ReadKey(); }運行結果:可以發現異步任務Task.Run()還沒有完成,但是因為cst.Cancel()運行,token的屬性IsCancellationRequested變為true,異步循環結束。
說明:取消一個異步操作的過程,注意,該過程是協同的。
即:調用CancellationTokenSource的Cancel時,它本身并不會執行取消操作。
而是會將CancellationToken的IsCancellationRequested屬性設置為true。
包含CancellationToken的代碼負責檢查該屬性,并判斷是否需要停止執行并返回。
?
?
4.并行Linq(PLinq)
?
4.1 AsParallel()
System.Linq名稱空間中有一個ParallelEnumerable類,該類中的方法可以分解Linq查詢的工作,使其分布在多個線程上,即實現并行查詢。
為并行運行而設計的LINQ查詢稱為PLINQ查詢。
下面讓我們先簡單的理一理:
首先我們都知道Enumerable類為IEnumberable<T>接口擴展了一系列的靜態方法。(就是我們使用Linq方法語法的中用的哪些常用的靜態方法,自行F12)
正如MSDN中所說:“ParallelEnumberable是Enumberable的并行等效項”,ParallelEnumberable類則是Enumerable類的并行版本,
F12查看定義可以看到ParallelEnumerable類中幾乎所有的方法都是對ParallelQuery<TSource>接口的擴展,
但是,在ParallelEnumberable類有一個重要的例外,AsParallel()?方法還對IEnumerable<T>接口的擴展,并且返回的是一個ParallelQuery<TSource>類型的對象,
所以呢?凡是實現類IEnumberable<T>集合可以通過調用靜態方法AsParallel(),返回一個ParallelQuery類型的對象,之后就可以使用ParallelEnumerable類中的異步版本的靜態查詢方法了!
注意在運行PLinq的時候,PLinq會自動的判斷如果查詢能從并行化中受益,則將同時運行。而如果并行執行查詢會損害性能,PLINQ將按順序運行查詢。
示例:求1到50000000中可以整除3的數,將所求的結果倒序存放在modThreeIsZero[]中
這是需要非常多的重復運算,所以我們可以對比按照一般Linq查詢下方式和PLinq查詢,對比一些需要的時間。
static void Main(string[] args) {int[] intArray = Enumerable.Range(1, 50000000).ToArray();Stopwatch sw = new Stopwatch();//順序查詢sw.Start();int[] modThreeIsZero1 = intArray.Select(n => n).Where(n => n % 3 == 0).OrderByDescending(n => n).ToArray();sw.Stop();Console.WriteLine($"順序查詢,運行時間:{sw.ElapsedMilliseconds}毫秒,可以整除3的個數:{modThreeIsZero1.Count()}");//使用AsParallel()實現并行查詢//AsParallel()方法返回ParallelQuery<TSourc>類型對象。因為返回的類型,所以編譯器選擇的Select()、Where()等方法是ParallelEnumerable.Where(),而不是Enumerable.Where()。sw.Restart();int[] modThreeIsZero2 = intArray.AsParallel().Select(n => n).Where(n => n % 3 == 0).OrderByDescending(n => n).ToArray();sw.Stop();Console.WriteLine($"并行查詢,運行時間:{sw.ElapsedMilliseconds}毫秒,可以整除3的個數:{modThreeIsZero2.Count()}");Console.ReadKey(); }說明:AsParallel()方法返回ParallelQuery<TSourc>類型對象。因為返回的類型,所以編譯器選擇的Select()、Where()等方法是ParallelEnumerable.Where(),而不是Enumerable.Where()。
運行結果:
可以對比結果,在大規模的Linq查詢中,同步查詢和并行查詢兩者的運行時間的差距還是很大的!
但是小規模的Linq查詢二者的效果其實并沒有很明顯。
4.2 取消并行查詢
在3.6取消異步操作中解釋了如何取消一個長時間的任務,
那么對于長時間運行的PLinq也是可以取消的
同樣是使用CancellationTokenSource生成一個CancellationToken對象作為token
怎么把token給PLinq呢?使用ParallelQuery<TSource>中靜態方法WithCancellation(token)
在PLinq中,若是取消了并行操作,則會拋出OperationCanceledException
示例:
static void Main(string[] args) {//具體的作用和含義可以看0030取消一個異步操作CancellationTokenSource cts = new CancellationTokenSource();CancellationToken ct = cts.Token;int[] intArray = Enumerable.Range(1, 50000000).ToArray();Task<int[]> task = Task.Run(() =>{try{int[] modThreeIsZero = intArray.AsParallel().WithCancellation(ct).Select(n => n).Where(n=> n% 3 == 0).OrderByDescending(n => n).ToArray();return modThreeIsZero;}catch (OperationCanceledException ex)//一旦PLinq中取消查詢就會觸發OperationCanceledException異常{Console.WriteLine(ex.Message);//注意:Message的內容就是:已取消該操作return null;}});Console.WriteLine("取消PLinq?Y/N");string input = Console.ReadLine();if (input.ToLower().Equals("y")){cts.Cancel();//取消并行查詢Console.WriteLine("取消了PLinq!");//undone:怎么驗證已經真的取消了}else{Console.WriteLine("Loading……");Console.WriteLine(task.Result.Count());}Console.ReadKey(); }
?
?
5.參考&源代碼下載
書籍:精通C#
書籍:C#高級編程
書籍:ASP.NET MVC5網站開發之美
文檔:.NET API 瀏覽器
點擊:源代碼下載
唉,書真是越看越厚,皆是淺嘗輒止,先到這里吧!
總結
以上是生活随笔為你收集整理的.NET异步程序设计之任务并行库的全部內容,希望文章能夠幫你解決所遇到的問題。
- 上一篇: ASP.NET Core真实管道详解[2
- 下一篇: 信用卡持卡人哪些行为会产生不良信用记录