歡迎來到Linux教程網
Linux教程網
Linux教程網
Linux教程網
您现在的位置: Linux教程網 >> UnixLinux >  >> Linux編程 >> Linux編程

細說.NET中的多線程:使用Task

上一節我們介紹了線程池相關的概念以及用法。我們可以發現ThreadPool. QueueUserWorkItem是一種起了線程之後就不管了的做法。但是實際應用過程,我們往往會有更多的需求,比如如果更簡單的知道線程池裡面的某些線程什麼時候結束,線程結束後如何執行別的任務。Task可以說是ThreadPool的升級版,在線程任務調度,並行編程中都有很大的作用。

創建並且初始化Task

使用lambda表達式創建Task

Task.Factory.StartNew(() => Console.WriteLine("Hello from a task!"));

var task = new Task(() => Console.Write("Hello"));

task.Start();

用默認參數的委托創建Task

 

using System;

using System.Threading.Tasks;

namespace MultiThread

{

    class ThreadTest

    {

        static void Main()

        {

            var task = Task.Factory.StartNew(state => Greet("Hello"), "Greeting");

            Console.WriteLine(task.AsyncState);  // Greeting

            task.Wait();

        }

        static void Greet(string message) { Console.Write(message); }

    }

}

這種方式的一個優點是,task.AsyncState作為一個內置的屬性,可以在不同線程中獲取參數的狀態。

System.Threading.Tasks.TaskCreateOptions

創建Task的時候,我們可以指定創建Task的一些相關選項。在.Net 4.0中,有如下選項:

LongRunning

用來表示這個Task是長期運行的,這個參數更適合block線程。LongRunning線程一般回收的周期會比較長,因此CLR可能不會把它放到線程池中進行管理。

PreferFairness

表示讓Task盡量以公平的方式運行,避免出現某些線程運行過快或者過慢的情況。

AttachedToParent

表示創建的Task是當前線程所在Task的子任務。這一個用途也很常見。

下面的代碼是創建子任務的示例:

using System;

using System.Threading;

using System.Threading.Tasks;

namespace MultiThread

{

    class ThreadTest

    {

        public static void Main(string[] args)

        {

            Task parent = Task.Factory.StartNew(() =>

            {

                Console.WriteLine("I am a parent");

                Task.Factory.StartNew(() =>        // Detached task

                {

                    Console.WriteLine("I am detached");

                });

                Task.Factory.StartNew(() =>        // Child task

                {

                    Console.WriteLine("I am a child");

                }, TaskCreationOptions.AttachedToParent);

            });

            parent.Wait();

            Console.ReadLine();

        }

    }

}

如果你等待你一個任務結束,你必須同時等待任務裡面的子任務結束。這一點很重要,尤其是你在使用Continue的時候。(後面會介紹)

等待Task

在ThreadPool內置的方法中無法實現的等待,在Task中可以很簡單的實現了:

using System;

using System.Threading;

using System.Threading.Tasks;

namespace MultiThread

{

    class ThreadTest

    {

        static void Main()

        {

            var t1 = Task.Run(() => Go(null));

            var t2 = Task.Run(() => Go(123));

            Task.WaitAll(t1, t2);//等待所有Task結束

            //Task.WaitAny(t1, t2);//等待任意Task結束

        }

        static void Go(object data)  // data will be null with the first call.

        {

            Thread.Sleep(5000);

            Console.WriteLine("Hello from the thread pool! " + data);

        }

    }

}

注意:

當你調用一個Wait方法時,當前的線程會被阻塞,直到Task返回。但是如果Task還沒有被執行,這個時候系統可能會用當前的線程來執行調用Task,而不是新建一個,這樣就不需要重新創建一個線程,並且阻塞當前線程。這種做法節省了創建新線程的開銷,也避免了一些線程的切換。但是也有缺點,當前線程如果和被調用的Task同時想要獲得一個lock,就會導致死鎖。

Task異常處理

當等待一個Task完成的時候(調用Wait或者或者訪問Result屬性的時候),Task任務中沒有處理的異常會被封裝成AggregateException重新拋出,InnerExceptions屬性封裝了各個Task沒有處理的異常。

using System;

using System.Threading.Tasks;

namespace MultiThreadTest

{

    class Program

    {

        static void Main(string[] args)

        {

            int x = 0;

            Task<int> calc = Task.Factory.StartNew(() => 7 / x);

            try

            {

                Console.WriteLine(calc.Result);

            }

            catch (AggregateException aex)

            {

                Console.Write(aex.InnerException.Message);  // Attempted to divide by 0

            }

        }

    }

}

對於有父子關系的Task,子任務未處理的異常會逐層傳遞到父Task,並且最後包裝在AggregateException中。

using System;

using System.Threading.Tasks;

namespace MultiThreadTest

{

    class Program

    {

        static void Main(string[] args)

        {

            TaskCreationOptions atp = TaskCreationOptions.AttachedToParent;

            var parent = Task.Factory.StartNew(() =>

            {

                Task.Factory.StartNew(() =>  // Child

                {

                    Task.Factory.StartNew(() => { throw null; }, atp);  // Grandchild

                }, atp);

            });

            // The following call throws a NullReferenceException (wrapped

            // in nested AggregateExceptions):

            parent.Wait();

        }

    }

}

取消Task

如果想要支持取消任務,那麼在創建Task的時候,需要傳入一個CancellationTokenSouce

示例代碼:

using System;

using System.Threading;

using System.Threading.Tasks;

namespace MultiThreadTest

{

    class Program

    {

        static void Main(string[] args)

        {

            var cancelSource = new CancellationTokenSource();

            CancellationToken token = cancelSource.Token;

            Task task = Task.Factory.StartNew(() =>

            {

                // Do some stuff...

                token.ThrowIfCancellationRequested();  // Check for cancellation request

                // Do some stuff...

            }, token);

            cancelSource.Cancel();

            try

            {

                task.Wait();

            }

            catch (AggregateException ex)

            {

                if (ex.InnerException is OperationCanceledException)

                    Console.Write("Task canceled!");

            }

            Console.ReadLine();

        }

    }

}

任務的連續執行

Continuations

任務調度也是常見的需求,Task支持一個任務結束之後執行另一個任務。

Task task1 = Task.Factory.StartNew(() => Console.Write("antecedant.."));

Task task2 = task1.ContinueWith(task =>Console.Write("..continuation"));

Continuations 和Task<TResult>

Task也有帶返回值的重載,示例代碼如下:

Task.Factory.StartNew<int>(() => 8)

    .ContinueWith(ant => ant.Result * 2)

    .ContinueWith(ant => Math.Sqrt(ant.Result))

    .ContinueWith(ant => Console.WriteLine(ant.Result));  // output 4

子任務

前面提到了,當你等待一個任務的時候,同時需要等待它的子任務���成。

下面代碼演示了帶子任務的Task:

using System;

using System.Threading.Tasks;

using System.Threading;

namespace MultiThreadTest

{

    class Program

    {

        public static void Main(string[] args)

        {

            Task<int[]> parentTask = Task.Factory.StartNew(() =>

            {

                int[] results = new int[3];

                Task t1 = new Task(() => { Thread.Sleep(3000); results[0] = 0; }, TaskCreationOptions.AttachedToParent);

                Task t2 = new Task(() => { Thread.Sleep(3000); results[1] = 1; }, TaskCreationOptions.AttachedToParent);

                Task t3 = new Task(() => { Thread.Sleep(3000); results[2] = 2; }, TaskCreationOptions.AttachedToParent);

                t1.Start();

                t2.Start();

                t3.Start();

                return results;

            });

            Task finalTask = parentTask.ContinueWith(parent =>

            {

                foreach (int result in parent.Result)

                {

                    Console.WriteLine(result);

                }

            });

            finalTask.Wait();

            Console.ReadLine();

        }

    }

}

這段代碼的輸出結果是: 1,2,3

FinalTask會等待所有子Task結束後再執行。

TaskFactory

關於TaskFactory,上面的例子中我們使用了System.Threading.Tasks .Task.Factory屬性來快速的創建Task。當然你也可以自己創建TaskFactory,你可以指定自己的TaskCreationOptions,TaskContinuationOptions來使得通過你的Factory創建的Task默認行為不同。

.Net中有一些默認的創建Task的方式,由於TaskFactory創建Task的默認行為不同可能會導致一些不容易發現的問題。

如在.NET 4.5中,Task加入了一個Run的靜態方法:

Task.Run(someAction);

如果你用這個方法代替上面例子中的Task.Factory.StartNew,就無法得到正確的結果。原因是Task.Run創建Task的行為默認是默認是拒絕添加子任務的。上面的代碼等價於:

    Task.Factory.StartNew(someAction, CancellationToken.None, TaskCreationOptions.DenyChildAttach, TaskScheduler.Default);

你也可以創建具有自己默認行為的TaskFactory。

無論ThreadPool也好,或者Task,微軟都是在想進辦法來實現線程的重用,來節省不停的創建銷毀線程帶來的開銷。線程池內部的實現可能在不同版本中有不同的機制。如果可能的話,使用線程池來管理線程仍然是建議的選擇。

我們主要介紹了一下Task的基本用法,在我們編程過程中,有一些使用Task來提升程序性能的場景往往是很相似的,微軟為了簡化編程,在System.Threading.Tasks.Parallel中封裝了一系列的並行類,內部也是通過Task來實現的。

Parallel的For,Foreach,Invoke 方法

在編程過程中,我們經常會用到循環語句:

for (int i = 0; i < 10; i++)

{

    DoSomeWork(i);

}

如果循環過程中的工作可以是並行的話,那麼我們可以用如下語句:

Parallel.For(0, 10, i => DoSomeWork(i));

我們也經常會使用Foreach來遍歷某個集合:

foreach (var item in collection)

{

    DoSomeWork(item);

}

如果我們用一個線程池來執行裡面的任務,那麼我們可以寫成:

Parallel.ForEach(collection, item => DoSomeWork(item));

最後,如果你想並行的執行幾個不同的方法,你可以:

Parallel.Invoke(Method1, Method2, Method3);

如果你看下後台的實現,你會發現基本都是基於Task的線程池,當然你也可以通過手動創建一個Task集合,然後等待所有的任務結束來實現同樣的功能。上面的Parallel.For和Parallel.Forach方法並不以為這你可以尋找你代碼裡面所有用到For和Foreach方法,並且替代他們,因為每一個任務都會分配一個委托,並且在線程池裡執行,如果委托裡面的任務是線程不安全的,你可能還需要lock來保證線程安全,使用lock本身就會造成性能上的損耗。如果每一個任務都是需要長時間執行並且線程安全的,Parallel會給你帶來不錯的性能提升。對於短任務,或者線程不安全的任務,你需要權衡下,你是否真的需要使用Parallel。

Copyright © Linux教程網 All Rights Reserved