除了上篇中提到的线程池，本篇介绍一种新的实现异步操作的方法--任务(Task)。
主要内容：
- 任务的介绍
- 任务的基本应用
- 子任务和任务工厂
- 任务调度器
- 并行任务Parallel
1. 任务的介绍
利用ThreadPool的QueueUserWorkItem方法建立的异步操作存在一些限制：
- 异步操作没有返回值
- 没有内建的机制来通知异步操作什么时候完成
而使用任务(Task)来建立异步操作可以克服上述限制，同时还解决了其他一些问题。
任务(Task)对象和线程池相比，多了很多状态字段和方法，便于更好的控制任务(Task)的运行。
当然，任务(Task)提供大量的功能也是有代价的，意味着更多的内存消耗。所以在实际使用中，如果不用任务(Task)的附加功能，那么就使用ThreadPool的QueueUserWorkItem方法。
通过任务的状态（TaskStatus），可以了解任务(Task)的生命周期。
TaskStatus是一个枚举类型，定义如下：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
|
public
enum
TaskStatus
{
// 运行前状态
Created = 0,
// 任务被显式创建,通过Start()开始这个任务
WaitingForActivation = 1,
// 任务被隐式创建,会自动开始
WaitingToRun = 2,
// 任务已经被调度,但是还没有运行
// 运行中状态
Running = 3,
// 任务正在运行
WaitingForChildrenToComplete = 4,
// 等待子任务完成
// 运行完成后状态
RanToCompletion = 5,
// 任务正常完成
Canceled = 6,
// 任务被取消
Faulted = 7,
// 任务出错
}
|
构造一个Task后，它的状态为Create。
启动后，状态变为WaitingToRun。
实际在一个线程上运行时，状态变为Running。
运行完成后，根据实际情况，状态变为RanToCompletiion，Canceled，Faulted三种中的一种。
如果Task不是通过new来创建的，而是通过以下某个函数创建的，那么它的状态就是WaitingForActivation：
ContinueWith，ContinueWhenAll，ContinueWhenAny，FromAsync。
如果Task是通过构造一个TaskCompletionSource<TResult>对象来创建的，该Task在创建时也是处于WaitingForActivation状态。
2. 任务的基本应用
下面演示任务的创建，取消，等待等基本使用方法。
2.1 创建并启动一个Task
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
|
using
System;
using
System.Threading.Tasks;
using
System.Threading;
public
class
CLRviaCSharp_19
{
static
void
Main(
string
[] args)
{
Console.WriteLine(
"Main Thread start!"
);
// 创建一个Task
Task t1 =
new
Task(() => {
Console.WriteLine(
"Task start"
);
Thread.Sleep(1000);
Console.WriteLine(
"Task end"
);
});
// 启动Task
t1.Start();
// 主线程并没有等待Task，在Task完成前就已经完成了
Console.WriteLine(
"Main Thread end!"
);
Console.ReadKey(
true
);
}
}
|
2.2 主线程等待子线程完成
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
|
using
System;
using
System.Threading.Tasks;
using
System.Threading;
public
class
CLRviaCSharp_19
{
static
void
Main(
string
[] args)
{
Console.WriteLine(
"Main Thread start!"
);
// 创建2个Task
Task t1 =
new
Task(() => {
Console.WriteLine(
"Task1 start"
);
Thread.Sleep(1000);
Console.WriteLine(
"Task1 end"
);
});
Task t2 =
new
Task(() =>
{
Console.WriteLine(
"Task2 start"
);
Thread.Sleep(2000);
Console.WriteLine(
"Task2 end"
);
});
// 启动Task
t1.Start();
t2.Start();
// 当t1和t2中任何一个完成后，主线程继续后面的操作
// Task.WaitAny(new Task[] { t1, t2 });
// 当t1和t2中全部完成后，主线程继续后面的操作
Task.WaitAll(
new
Task[] { t1, t2 });
Console.WriteLine(
"Main Thread end!"
);
Console.ReadKey(
true
);
}
}
|
等待的方法WaitAll和WaitAny可根据应用场景选用一个。
2.3 取消Task
取消Task和取消一个线程类似，使用CancellationTokenSource。
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
|
using
System;
using
System.Threading.Tasks;
using
System.Threading;
public
class
CLRviaCSharp_19
{
static
void
Main(
string
[] args)
{
Console.WriteLine(
"Main Thread start!"
);
CancellationTokenSource cts =
new
CancellationTokenSource();
// 创建2个Task
Task t1 =
new
Task(() => {
Console.WriteLine(
"Task1 start"
);
for
(
int
i = 0; i < 100; i++)
{
if
(!cts.Token.IsCancellationRequested)
{
Console.WriteLine(
"Count : "
+ i.ToString());
Thread.Sleep(1000);
}
else
{
Console.WriteLine(
"Task1 is Cancelled!"
);
break
;
}
}
Console.WriteLine(
"Task1 end"
);
}, cts.Token);
// 启动Task
t1.Start();
Thread.Sleep(3000);
// 运行3秒后取消Task
cts.Cancel();
// 为了测试取消操作，主线程等待Task完成
Task.WaitAny(
new
Task[] { t1 });
Console.WriteLine(
"Main Thread end!"
);
Console.ReadKey(
true
);
}
}
|
3. 子任务和任务工厂
3.1 延续任务
为了保证程序的伸缩性，应该尽量避免线程阻塞，这就意味着我们在等待一个任务完成时，最好不要用Wait，而是让一个任务结束后自动启动它的下一个任务。
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
|
using
System;
using
System.Threading.Tasks;
using
System.Threading;
public
class
CLRviaCSharp_19
{
static
void
Main(
string
[] args)
{
Console.WriteLine(
"Main Thread start!"
);
// 第一个Task
Task<
int
> t1 =
new
Task<
int
>(() =>
{
Console.WriteLine(
"Task 1 start!"
);
Thread.Sleep(2000);
Console.WriteLine(
"Task 1 end!"
);
return
1;
});
// 启动第一个Task
t1.Start();
// 因为TaskContinuationOptions.OnlyOnRanToCompletion，
// 所以第一个Task正常结束时,启动第二个Task。
// TaskContinuationOptions.OnlyOnFaulted，则第一个Task出现异常时,启动第二个Task
// 其他可详细参考TaskContinuationOptions定义的各个标志
t1.ContinueWith(AnotherTask, TaskContinuationOptions.OnlyOnRanToCompletion);
Console.WriteLine(
"Main Thread end!"
);
Console.ReadKey(
true
);
}
// 第二个Task的处理都在AnotherTask函数中，
// 第二个Task的引用其实就是上面ContinueWith函数的返回值。
// 这里没有保存第二个Task的引用
private
static
void
AnotherTask(Task<
int
> task)
{
Console.WriteLine(
"Task 2 start!"
);
Thread.Sleep(1000);
Console.WriteLine(
"Task 1's return Value is : "
+ task.Result);
Console.WriteLine(
"Task 2 end!"
);
}
}
|
3.2 子任务
定义子任务时，注意一定要加上TaskCreationOptions.AttachedToParent，这样父任务会等待子任务执行完后才结束。
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
|
using
System;
using
System.Threading.Tasks;
using
System.Threading;
public
class
CLRviaCSharp_19
{
static
void
Main(
string
[] args)
{
Console.WriteLine(
"Main Thread start!"
);
Task<
int
[]> parentTask =
new
Task<
int
[]>(() =>
{
var
result =
new
int
[3];
// 子任务1
new
Task(() => {
Console.WriteLine(
"sub task 1 start!"
);
Thread.Sleep(1000);
Console.WriteLine(
"sub task 1 end!"
);
result[0] = 1;
}, TaskCreationOptions.AttachedToParent).Start();
// 子任务2
new
Task(() =>
{
Console.WriteLine(
"sub task 2 start!"
);
Thread.Sleep(1000);
Console.WriteLine(
"sub task 2 end!"
);
result[1] = 2;
}, TaskCreationOptions.AttachedToParent).Start();
// 子任务3
new
Task(() =>
{
Console.WriteLine(
"sub task 3 start!"
);
Thread.Sleep(1000);
Console.WriteLine(
"sub task 3 end!"
);
result[2] = 3;
}, TaskCreationOptions.AttachedToParent).Start();
return
result;
});
parentTask.Start();
Console.WriteLine(
"Parent Task's Result is :"
);
foreach
(
int
result
in
parentTask.Result)
Console.Write(
"{0}\t"
, result);
Console.WriteLine();
Console.WriteLine(
"Main Thread end!"
);
Console.ReadKey(
true
);
}
}
|
上面的例子中，可以把TaskCreationOptions.AttachedToParent删掉试试，打印出来的Result应该是3个0，而不是1 2 3。
3个子任务的执行顺序也和定义的顺序无关，比如任务3可能最先执行（与CPU的调度有关）。
3.3 任务工厂
除了上面的方法，还可以使用任务工厂来批量创建任务。
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
|
using
System;
using
System.Threading.Tasks;
using
System.Threading;
public
class
CLRviaCSharp_19
{
static
void
Main(
string
[] args)
{
Console.WriteLine(
"Main Thread start!"
);
Task<
int
[]> parentTask =
new
Task<
int
[]>(() =>
{
var
result =
new
int
[3];
TaskFactory tf =
new
TaskFactory(TaskCreationOptions.AttachedToParent, TaskContinuationOptions.None);
// 子任务1
tf.StartNew(() =>
{
Console.WriteLine(
"sub task 1 start!"
);
Thread.Sleep(1000);
Console.WriteLine(
"sub task 1 end!"
);
result[0] = 1;
});
// 子任务2
tf.StartNew(() =>
{
Console.WriteLine(
"sub task 2 start!"
);
Thread.Sleep(1000);
Console.WriteLine(
"sub task 2 end!"
);
result[1] = 2;
});
// 子任务3
tf.StartNew(() =>
{
Console.WriteLine(
"sub task 3 start!"
);
Thread.Sleep(1000);
Console.WriteLine(
"sub task 3 end!"
);
result[2] = 3;
});
return
result;
});
parentTask.Start();
Console.WriteLine(
"Parent Task's Result is :"
);
foreach
(
int
result
in
parentTask.Result)
Console.Write(
"{0}\t"
, result);
Console.WriteLine();
Console.WriteLine(
"Main Thread end!"
);
Console.ReadKey(
true
);
}
}
|
使用任务工厂与上面3.2中直接定义子任务相比，优势主要在于可以共享子任务的设置，比如在TaskFactory中设置了TaskCreationOptions.AttachedToParent，那么它启动的子任务都具有这个属性了。
当然，任务工厂(TaskFactory)还提供了很多控制子任务的函数，用的时候可以看看它的类定义。
4. 任务调度器
上面例子中任务的各种操作(运行，等待，取消等等)，都是由CLR的任务调度器来调度的。
FCL公开了2种任务调度器：线程池任务调度器和同步上下文任务调度器。
默认情况下，应用程序都是使用的线程池任务调度器。WPF和Winform中通常使用同步上下文任务调度器。
CLR的任务调度器类(TaskScheduler)中有个Default属性返回的就是线程池任务调度器。
还有个FromCurrentSynchronizationContext方法，返回的是同步上下文任务调度器。
我们也可以通过继承CLR中的任务调度器(TaskScheduler)来定制适合自己业务需要的任务调度器。
下面我们定制一个简单的TaskScheduler，将3.3中每个子任务的打印信息的功能移到自定义的任务调度器MyTaskScheduler中。
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
|
using
System;
using
System.Threading.Tasks;
using
System.Threading;
using
System.Collections.Generic;
public
class
CLRviaCSharp_19
{
static
void
Main(
string
[] args)
{
Console.WriteLine(
"Main Thread start!"
);
Task<
int
[]> parentTask =
new
Task<
int
[]>(() =>
{
var
result =
new
int
[3];
// 这里的TaskFactory中指定的是自定义的任务调度器MyTaskScheduler
TaskFactory tf =
new
TaskFactory(CancellationToken.None, TaskCreationOptions.AttachedToParent,
TaskContinuationOptions.None,
new
MyTaskScheduler());
// 子任务1
tf.StartNew(() =>
{
Thread.Sleep(1000);
result[0] = 1;
});
// 子任务2
tf.StartNew(() =>
{
Thread.Sleep(1000);
result[1] = 2;
});
// 子任务3
tf.StartNew(() =>
{
Thread.Sleep(1000);
result[2] = 3;
});
return
result;
});
parentTask.Start();
Console.WriteLine(
"Parent Task's Result is :"
);
foreach
(
int
result
in
parentTask.Result)
Console.Write(
"{0}\t"
, result);
Console.WriteLine();
Console.WriteLine(
"Main Thread end!"
);
Console.ReadKey(
true
);
}
}
// 自定义的TaskScheduler，没什么实际的作用，只是为了实验自定义TaskScheduler
public
class
MyTaskScheduler : TaskScheduler
{
private
IList<Task> _lstTasks;
public
MyTaskScheduler()
{
_lstTasks =
new
List<Task>();
}
#region inherit from TaskScheduler
protected
override
System.Collections.Generic.IEnumerable<Task> GetScheduledTasks()
{
return
_lstTasks;
}
protected
override
void
QueueTask(Task task)
{
_lstTasks.Add(task);
// 将原先的打印信息，移到此处统一处理
Console.WriteLine(
"task "
+ task.Id +
" is start!"
);
TryExecuteTask(task);
Console.WriteLine(
"task "
+ task.Id +
" is end!"
);
}
protected
override
bool
TryExecuteTaskInline(Task task,
bool
taskWasPreviouslyQueued)
{
return
TryExecuteTask(task);
}
#endregion
}
|
5. 并行任务Parallel
Parallel是为了简化任务编程而新增的静态类，利用Parallel可以将平时的循环操作都并行起来。
下例演示了for并行循环，foreach并行循环与之类似。
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
|
using
System;
using
System.Threading.Tasks;
using
System.Threading;
using
System.Diagnostics;
public
class
CLRviaCSharp_19
{
static
void
Main(
string
[] args)
{
Console.WriteLine(
"Main Thread start!"
);
int
max = 10;
// 普通循环
long
start = Stopwatch.GetTimestamp();
for
(
int
i = 0; i < max; i++)
{
Thread.Sleep(1000);
}
Console.WriteLine(
"{0:N0}"
, Stopwatch.GetTimestamp() - start);
// 并行的循环
start = Stopwatch.GetTimestamp();
Parallel.For(0, max, i => { Thread.Sleep(1000); });
Console.WriteLine(
"{0:N0}"
, Stopwatch.GetTimestamp() - start);
Console.WriteLine(
"Main Thread end!"
);
Console.ReadKey(
true
);
}
}
|
在上面的例子中，采用并行循环消耗的时间不到原先的一半。
但是，采用并行循环需要满足一个条件，就是for循环中的内容能够并行才行。
比如for循环中是个对 循环变量i 进行的累加操作(例如sum += i;)，那就不能使用并行循环。
还有一点需要注意，Parallel的方法本身有开销。
所以如果for循环内的处理比较简单的话，那么直接用for循环可能更快一些。
比如将上例中的Thread.Sleep(1000);删掉，再运行程序发现，直接for循环要快很多。
本文转自wang_yb博客园博客，原文链接：http://www.cnblogs.com/wang_yb/archive/2011/11/10/2244745.html，如需转载请自行联系原作者
