.NET并發(fā)編程-任務(wù)函數(shù)并行

本小節(jié)介紹一種簡(jiǎn)單的函數(shù)組合來并行執(zhí)行任務(wù)方式，達(dá)到不阻塞程序提高性能的目的。

1、任務(wù)并行

回顧下什么是任務(wù)并行，任務(wù)并行是在相同或不同的數(shù)據(jù)集上用時(shí)執(zhí)行多個(gè)不同的函數(shù)，區(qū)別于數(shù)據(jù)并行是在數(shù)據(jù)集的元素之間同時(shí)執(zhí)行同一個(gè)函數(shù)。

生產(chǎn)中可能會(huì)涉及不同的任務(wù)函數(shù)，處理不同的復(fù)雜的結(jié)構(gòu)數(shù)據(jù)，通過利用在.NET提供的一些模型工具箱我們可以較為簡(jiǎn)便的任務(wù)并行跑起來。

2、.NET中的任務(wù)并行化支持

由淺入深，.NET1.0開始就提供線程的訪問控制。System.Thread，可以代碼控制創(chuàng)建啟動(dòng)銷毀現(xiàn)場(chǎng)。但線程的創(chuàng)建開銷比較大，后面有提供了ThreadPool類，線程池有助于克服性能問題。在初始化期間就加載了一組線程，然后重用這些線程，避免了頻繁創(chuàng)建銷毀線程的開銷。

Action<string> downloadSite = url => {    var content = new WebClient().DownloadString(url);    Console.WriteLine($"The size of the web site {url} is {content.Length}");}; 
var threadA = new Thread(() => downloadSite("http://www.nasdaq.com"));var threadB = new Thread(() => downloadSite("http://www.bbc.com"));
threadA.Start();threadB.Start(); threadA.Join();threadB.Join();  
ThreadPool.QueueUserWorkItem(o => downloadSite("http://www.nasdaq.com"));ThreadPool.QueueUserWorkItem(o => downloadSite("http://www.bbc.com"));

像上面所示傳統(tǒng)的方式也很繁瑣，而且有很多弊端，比如無法獲取結(jié)果，沒有內(nèi)置通知等。因?yàn)橛痔峁┝薚PL任務(wù)并行庫(kù)。

3、.NET任務(wù)并行庫(kù)

TPL在ThreadPool上實(shí)現(xiàn)了很多優(yōu)化，簡(jiǎn)化了添加并行的過程，通過Task對(duì)象提供支持，以取消和管理狀態(tài)，處理和傳播異常，以及控制工程線程的執(zhí)行。

TPL提供很多種調(diào)度任務(wù)的方式，Invoke是最簡(jiǎn)單的一種。類似的還有Parallel.ForEach

System.Threading.Tasks.Parallel.Invoke(        Action(fun () -> convertImageTo3D (pathCombine "MonaLisa.jpg") (pathCombine "MonaLisa3D.jpg")),        Action(fun () -> setGrayscale (pathCombine "LadyErmine.jpg") (pathCombine "LadyErmineRed.jpg")),        Action(fun () -> setRedscale (pathCombine "GinevraBenci.jpg") (pathCombine "GinevraBenciGray.jpg")))

此方法接收任意數(shù)量的action委托參數(shù)，并為每一個(gè)委托創(chuàng)建任務(wù)。但是，action委托沒有輸入?yún)?shù)，并且返回void，這樣的函數(shù)會(huì)有副作用。當(dāng)所有任務(wù)終止時(shí)，Invoke方法將控制權(quán)交回給主線程以繼續(xù)執(zhí)行后續(xù)流程。在并行執(zhí)行獨(dú)立的異構(gòu)任務(wù)時(shí)，就是針對(duì)不同的結(jié)構(gòu)數(shù)據(jù)，此方法挺有效的。

弊端也很明顯，沒有輸入類型，返回為Void，也就限制了組合使用，執(zhí)行順序也無法保障。

4、C#void問題

和Null類似，Void也是一個(gè)頭疼的問題。函數(shù)式編程語言中每一個(gè)函數(shù)都有返回值，包括與void類似情況的unit類型，但是與void不同的是該值被視為一個(gè)值，概念上與bool和int沒多大區(qū)別。

unit是缺少其他特定值的表達(dá)式的類型，像打印日志到控制臺(tái)，寫入文件等，沒有特定的內(nèi)容需要返回，因?yàn)楹瘮?shù)需要返回unit。unit就是C#的void在F#中的等價(jià)產(chǎn)物。

在FP的函數(shù)就是一個(gè)映射，一個(gè)輸入映射一個(gè)輸出，這樣函數(shù)才是無副作用的。在命令式編程語言中丟失了這個(gè)概念。

可以參考F#unit自定義個(gè)C#中的unit

public struct Unit : IEquatable<Unit> {    public static readonly Unit Default = new Unit();  
    public override int GetHashCode() => 0;        public override bool Equals(object obj) => obj is Unit; 
    public override string ToString() => "()";
    public bool Equals(Unit other) => true;      public static bool operator ==(Unit lhs, Unit rhs) => true;     public static bool operator !=(Unit lhs, Unit rhs) => false; }

這樣可以讓每個(gè)函數(shù)都有返回值來確認(rèn)函數(shù)已完成，并且任何使用action委托的地方都可以使用func代替，只需要給func執(zhí)行返回值為unit即可。return Unit.Default;

5、延續(xù)傳遞風(fēng)格CPS

一種更新更好的機(jī)制是將剩余的計(jì)算傳遞給（在線程完成執(zhí)行后運(yùn)行的）回調(diào)函數(shù)以繼續(xù)工作。這種技術(shù)在FP中被稱為延續(xù)傳遞風(fēng)格Continuation-Passing Style CPS。通過將當(dāng)前函數(shù)的結(jié)果傳遞給下一個(gè)函數(shù)，以延續(xù)的形式為你提供執(zhí)行控制。

.NET中Task類提供比Thread更高級(jí)別的抽象，以便于控制每個(gè)每個(gè)任務(wù)操作的生命周期。

Task monaLisaTask = Task.Factory.StartNew(() => convertImageTo3D("MonaLisa.jpg", "MonaLisa3D.jpg"));Task ladyErineTask = new Task(() => setGrayscale("ladyErine.jpg", "ladyErine3D.jpg"));ladyErineTask.Start();Task ginevraBenciTask = Task.Run(() => setRedscale("ginevraBenci.jpg", "ginevraBenci3D.jpg"));

Task提供三種直接創(chuàng)建任務(wù)的方式，new Task方式可以控制在何處Start啟動(dòng)任務(wù)。

通過Task的ContinueWith可以延續(xù)任務(wù)。FromCurrentSynchronizationContext捕獲當(dāng)前不同上下文中運(yùn)行，如果需要同步UI請(qǐng)使用，會(huì)自動(dòng)選擇合適的上下文去更新。

Task ginevraBenciTask = Task.Run<Bitmap>(() => setRedscale("ginevraBenci.jpg", "ginevraBenci3D.jpg"));
ginevraBenciTask.ContinueWith(bitmap => {
    var bitmapImage = bitmap.Result; 
}, TaskScheduler.FromCurrentSynchronizationContext());

6、組合策略

使用ContinueWith可以延續(xù)任務(wù)，但較多的延續(xù)，代碼將比較繁瑣，而且如果要添加錯(cuò)誤處理或取消支持就不好添加了。所以要使用到函數(shù)閉包中說到的函數(shù)組合。

C#實(shí)現(xiàn)組合Compose函數(shù)如下

Func<A,C> Compose<A,B,C>(this Func<A.B> f ,Func<B,C> g)=>(n)=>g(f(n))

在并行Task中，f，g應(yīng)該是獨(dú)立運(yùn)行的，當(dāng)做兩個(gè)任務(wù)，f任務(wù)返回Task(B)，g任務(wù)返回Task(C)，所以改造如下

Func<A,Task<C>> Compose<A,B,C>(this Func<A.Task<B>> f ,Func<B,Task<C>> g)=>(n)=>g(f(n))

但是有問題的，f(n)返回類型Task(B)，無法直接給函數(shù)g使用，輸入類型不一致。

這個(gè)使用需要使用FP中常見的一種模式，單子Monad。對(duì)于命令式編程語言的程序員來說，壓根沒聽過啊。其實(shí)也是一種設(shè)計(jì)模式，就像裝飾器和適配器一樣。單子是一種數(shù)學(xué)模式，它通過封裝程序邏輯，保持函數(shù)式的純粹性以及提供一個(gè)強(qiáng)大的組合工具以組合使用提供類型的計(jì)算來控制副作用的執(zhí)行。比較晦澀難懂，還需要多看看官方文檔才行。

 我們定義一個(gè)Bind來提升類型，包裝B，然后組合就像下面這樣了。

static Task<C> Bind<B, C>(Task<B> b, Func<B, Task<C>> g){    return g(b.Result);}

Func<A,Task<C>> Compose<A,B,C>(this Func<A.Task<B>> f ,Func<B,Task<C>> g)=>(n)=>bind(f(n),g)