C#中的任务Task
在C#编程中,实现并行可以直接使用线程,但使用起来很繁琐;也可以使用线程池,线程池很大程度上简化了线程的使用,但是也有着一些局限,比如我们不知道作业什么时候完成,也取不到作业的返回值;解决线程池局限性的方案是使用任务
。本文将总结C#中Task
的使用。
类似于线程池工作项对异步操作的封装,任务是对异步操作的另一种形式的封装,这种封装抽象层次更高,让我们能够对异步操作进行更多的控制。
任务启动后,通过任务调度器TaskScheduler
来调度。.NET中提供两种任务调度器,一种是线程池任务调度器,也是默认调度器,它会将任务派发给线程池工作者线程;另一种是上下文同步任务调度器,它会将任务派发给当前上下文线程,例如GUI线程。此外,我们也能自定义任务调度器,例如可以将异步IO任务派发给线程池IO线程。
Task的使用方法
隐式使用
Parallel
静态类除了提供并行循环的各种重载,还提供了一个方法Parallel.Invoke
。这个方法可以创建并执行一个或多个异步任务,使用方法如下:
private static void DoWork(int workId = 0)
{
Console.WriteLine($"{DateTime.Now}=> Thread[{Thread.CurrentThread.ManagedThreadId}] started work[{workId}].");
Thread.Sleep(3000);
Console.WriteLine($"{DateTime.Now}=> Thread[{Thread.CurrentThread.ManagedThreadId}] done work[{workId}].");
}
public static void ImplicitUsingOfTask()
{
Parallel.Invoke(()=>DoWork(1),()=>DoWork(2),() => DoWork(3));
}
上例的运行结果如下:
2019/3/27 20:40:18=> Thread[9] started work[1].
2019/3/27 20:40:18=> Thread[12] started work[3].
2019/3/27 20:40:18=> Thread[10] started work[2].
2019/3/27 20:40:21=> Thread[9] done work[1].
2019/3/27 20:40:21=> Thread[12] done work[3].
2019/3/27 20:40:21=> Thread[10] done work[2].
对于简单的多任务并行,使用上述的方式很方便,但是这种方式与线程池一样,我们不能控制任务的执行或者获取任务返回值。
显式使用
相对于使用Parallel.Invoke
执行并行操作,更常用的是使用Task
和Task<T>
提供的方法进行异步和并行处理。下面是任务最基本的使用:
Task.Run(() =>
{
});
Task.Factory.StartNew(() =>
{
});
任务的常用操作
获取任务的返回值
具有返回值的任务使用Task<T>
,T
可根据我们的需求指定,下面是获取任务返回值的方法。
Task<int> task = Task<int>.Factory.StartNew(() =>
{
Thread.Sleep(3000);
return DateTime.Now.Day;
});
int day = task.Result;
需要说明的是,获取任务的结果会阻塞当前线程。
等待任务完成
有时候,我们需要等待一些任务全部完成后才能执行后续操作,有时候只要多个任务中的一个完成了,就可以执行后续操作。Task
提供了Wait
、WaitAll
和WaitAny
等方法满足我们的需求。下面的例子展示了各种等待方法的使用。
public static void TaskWait()
{
Stopwatch watch = new Stopwatch();
#region 场景1:等待一个任务完成
Task task = Task.Run(() => DoWorkOfTask(1000));
Console.WriteLine("start wait. work duration: 1000");
watch.Start();
task.Wait();
watch.Stop();
Console.WriteLine($"end wait. time: {watch.ElapsedMilliseconds}");
#endregion
#region 场景2:等待多个任务完成
Task[] tasks = new Task[3]
{
Task.Run(() => DoWorkOfTask(1000)),
Task.Run(() => DoWorkOfTask(2000)),
Task.Run(() => DoWorkOfTask(3000)),
};
Console.WriteLine("start wait all. work duration: min 1000 max 3000.");
watch.Restart();
Task.WaitAll(tasks);
watch.Stop();
Console.WriteLine($"end wait. time: {watch.ElapsedMilliseconds}");
#endregion
#region 场景3:等待某个任务完成
tasks = new Task[3]
{
Task.Run(() => DoWorkOfTask(1000)),
Task.Run(() => DoWorkOfTask(2000)),
Task.Run(() => DoWorkOfTask(3000)),
};
Console.WriteLine("start wait any. work duration: min 1000 max 3000.");
watch.Restart();
Task.WaitAny(tasks);
watch.Stop();
Console.WriteLine($"end wait. time: {watch.ElapsedMilliseconds}");
#endregion
}
private static void DoWorkOfTask(int workDuration)
{
Console.WriteLine($"{DateTime.Now}=> Thread[{Thread.CurrentThread.ManagedThreadId}] started task[{Task.CurrentId}].");
Thread.Sleep(workDuration);
Console.WriteLine($"{DateTime.Now}=> Thread[{Thread.CurrentThread.ManagedThreadId}] completed task[{Task.CurrentId}].");
}
使用Wait
、WaitAll
和WaitAny
方法时,我们可以设置超时时间或者传入取消Token,以控制等待时间。但这些方法返回布尔值,只能表明是否等待成功;假如我们需要知道所等待的任务返回值,则可以使用WhenAll
或WhenAny
方法,这两个方法不能控制等待时间,但会返回一个完成的任务。如下例:
Task<int>[] tasks = new Task<int>[3]
{
Task<int>.Factory.StartNew(() =>
{
Console.WriteLine($"task #{Task.CurrentId} run");
Thread.Sleep(100);
Console.WriteLine($"task #{Task.CurrentId} done");
return 100;
}),
Task<int>.Factory.StartNew(() =>
{
Console.WriteLine($"task #{Task.CurrentId} run");
Thread.Sleep(500);
Console.WriteLine($"task #{Task.CurrentId} done");
return 1000;
}),
Task<int>.Factory.StartNew(() =>
{
Console.WriteLine($"task #{Task.CurrentId} run");
Thread.Sleep(1000);
Console.WriteLine($"task #{Task.CurrentId} done");
return 10000;
}),
};
Task<int> task = Task.WhenAny(tasks).Result;
Console.WriteLine($"task #{task.Id}. result {task.Result}");
Task.WhenAll
和Task.WhenAny
在等待结束时,都会创建一个完成状态的任务,WhenAll
将等待的所有已完成任务的结果放入创建任务的结果中,WhenAny
则将等待的已完成任务放到创建任务的结果中。
任务延续
有时候,我们需要在一个任务完成时开始另一个任务。对于这种需求,我们可以使用Task
的ContinueWith
等方法来处理。
Task task = Task.Run(() => DoWorkOfTask(3000));
task.ContinueWith(t => DoWorkOfTask(1000));
运行结果:
2019/3/27 21:25:09=> Thread[10] started task[1].
2019/3/27 21:25:12=> Thread[10] completed task[1].
2019/3/27 21:25:12=> Thread[11] started task[2].
2019/3/27 21:25:13=> Thread[11] completed task[2].
我们还可以通过TaskContinuationOptions
指定延续任务的执行条件,如任务取消时或者任务出现异常时才执行,等。
子任务的使用
有时候,我们要在一个任务里面创建一些其他任务,并且还要在任务里面等待创建的任务完成,此时我们可以使用子任务。
Task parent = Task.Factory.StartNew(() =>
{
Console.WriteLine($"parent task #{Task.CurrentId} run.");
for (int i = 0; i < 10; i++)
{
Task.Factory.StartNew(() =>
{
Console.WriteLine($"child task #{Task.CurrentId} run.");
Thread.Sleep(1000);
Console.WriteLine($"child task #{Task.CurrentId} done.");
}, TaskCreationOptions.AttachedToParent);
}
});
parent.Wait();
Console.WriteLine($"parent task #{parent.Id} done.");
在一个任务中创建的新任务,默认情况下与父级任务是分离的,各自的运行不受影响,除非在创建任务时显式附加到父级任务中。例如,上例中如果不指定TaskCreationOptions.AttachedToParent
,parent.Wait()
就不会持续到所有子任务都执行完成。
任务的取消
我们在启动任务时,传入取消令牌CancellationToken
,当收到取消请求时,抛出取消异常并在等待任务完成时捕获异常TaskCanceledException
。我们通过这种方式控制任务的取消。
public static void TaskCancle()
{
Console.WriteLine("Press any key to begin. Press 'c' to cancel. ");
Console.ReadKey(true);
Console.WriteLine();
CancellationTokenSource tokenSource = new CancellationTokenSource();
ConcurrentBag<Task> tasks = new ConcurrentBag<Task>();
Task task1 = Task.Factory.StartNew(() => DoWorkOfTask(5000, tokenSource.Token), tokenSource.Token);
tasks.Add(task1);
Task task2 = Task.Factory.StartNew(() =>
{
for (int i = 0; i < 10; i++)
{
int duration = 1000 * i;
tasks.Add(Task.Factory.StartNew(()=>DoWorkOfTask(duration, tokenSource.Token), tokenSource.Token));
}
DoWorkOfTask(5000,tokenSource.Token);
}, tokenSource.Token);
tasks.Add(task2);
char ch = Console.ReadKey().KeyChar;
if (ch == 'c' || ch == 'C')
{
tokenSource.Cancel();
Console.WriteLine($"{DateTime.Now}=> Task cancellation requested.");
}
try
{
Task.WaitAll(tasks.ToArray());
}
catch (AggregateException ae)
{
foreach (Exception ex in ae.InnerExceptions)
{
TaskCanceledException tce = ex as TaskCanceledException;
string cancelledTask = tce == null ? string.Empty : $"Task #{tce.Task.Id}";
Console.WriteLine($"Exception: {ex.GetType().Name}. {cancelledTask}");
}
}
finally
{
tokenSource.Dispose();
}
Console.WriteLine();
foreach (Task task in tasks)
{
Console.WriteLine($"Task: #{task.Id} now is {task.Status}");
}
}
private static void DoWorkOfTask(int workDuration, CancellationToken cancleToken)
{
if (cancleToken.IsCancellationRequested)
{
Console.WriteLine($"{DateTime.Now}=> Task #{Task.CurrentId} was cancelled before it got started.");
cancleToken.ThrowIfCancellationRequested();
}
Console.WriteLine($"{DateTime.Now}=> Thread[{Thread.CurrentThread.ManagedThreadId}] started task #{Task.CurrentId}.");
Thread.Sleep(workDuration);
if (cancleToken.IsCancellationRequested)
{
Console.WriteLine($"{DateTime.Now}=> Task #{Task.CurrentId} was cancelled.");
cancleToken.ThrowIfCancellationRequested();
}
Console.WriteLine($"{DateTime.Now}=> Thread[{Thread.CurrentThread.ManagedThreadId}] completed task #{Task.CurrentId}.");
}
任务的异常处理
上面提到通过取消令牌抛出TaskCanceledException
的方式控制任务的取消,实际上,Task会把自身执行过程中的所有异常都包装到一个AggregateException
中,并传回调用线程。我们在主线程中通过捕获AggregateException
来进行异常处理。
简单的处理方式
我们可以在任务的调用线程捕获并遍历AggregateException
的内部异常,或者使用AggregateException
提供的Handle方法进行处理,如下:
Task task = Task.Run(() =>
{
throw new Exception($"Task #{Task.CurrentId} thrown an exception");
});
try
{
task.Wait();
}
catch (AggregateException ae)
{
foreach (Exception ex in ae.InnerExceptions)
{
Console.WriteLine($"foreach: {ex.Message}");
}
ae.Handle(ex=>
{
Console.WriteLine($"handle: {ex.Message}");
return true ;
});
}
使用延续任务处理任务的异常
有时候,我们可以给任务附加一个任务异常时才会执行的延续任务,并在延续任务中进行异常处理。
Task.Run(() => { throw new Exception($"Task #{Task.CurrentId} thrown an exception"); })
.ContinueWith(t =>
{
Console.WriteLine($"{t.Exception?.InnerException?.Message}");
}, TaskContinuationOptions.OnlyOnFaulted);
嵌套任务的异常处理
下面是一个3层嵌套的任务。
Task parent = Task.Factory.StartNew(() =>
{
for (int i = 0; i < 10; i++)
{
Task.Factory.StartNew(() =>
{
for (int j = 0; j < 10; j++)
{
Task.Factory.StartNew(() =>
{
throw new Exception($"Task #{Task.CurrentId} thrown an exception. ");
});
}
throw new Exception($"Task #{Task.CurrentId} thrown an exception. ");
});
}
throw new Exception($"Task #{Task.CurrentId} thrown an exception. ");
});
try
{
parent.Wait();
}
catch (AggregateException ae)
{
ae.Flatten().Handle(ex =>
{
Console.WriteLine(ex.Message);
return true;
});
}
运行上面的代码只会得到一行输出:
Task #1 thrown an exception.
看起来有点奇怪,为什么只捕获到一个异常呢?其实也是在情理之中的:任务默认只会把自身异常传递到它自己的调用线程,子任务是在父任务中调用的,其异常只会传递到父任务的执行线程,所以我们在父任务的调用线程,也就是我们的主线程中是捕获不到子任务的异常的。
取消上面代码的两处/*, TaskCreationOptions.AttachedToParent*/
,就会捕获到所有异常。
任务调度器
.NET提供的任务调度器
任务是由TaskScheduler
调度的,启动任务时,默认使用线程池任务调度器,任务将会被派发到线程池工作线程。线程池的调度前面已经总结过,这里不再展开。.NET提供的另一种任务调度器是同步上下文调度器,用TaskScheduler.FromCurrentSynchronizationContext()
获取,这个调度器会把任务派发给当前的上下文线程,常用在GUI应用程序中。
例如,我们在一个窗体中新建一个ListBox,新建几个任务向其中添加项,代码如下:
this.lbxMsg.Items.Add($"{DateTime.Now:O}=>Current thread is thread #{Thread.CurrentThread.ManagedThreadId} .");
for (int i = 0; i < 10; i++)
{
new Task(() =>
{
for (int j = 0; j < 3; j++)
{
this.lbxMsg.Items.Add($"{DateTime.Now:O}=> Task #{Task.CurrentId} add an item with thread #{Thread.CurrentThread.ManagedThreadId}.");
}
}).Start(TaskScheduler.FromCurrentSynchronizationContext());
}
运行上面的代码可以发现创建的任务都是由界面线程执行的。这里如果使用默认的任务调度器将产生"线程间操作无效"的异常。
实际使用时,可以给一个异步任务添加延续任务,来处理异步任务的结果或者异常等。如下:
Task.Run(() =>
{
Thread.Sleep(3000);
return 1000;
}).ContinueWith(t =>
{
this.lbxMsg.Items.Add(t.Result);
}, TaskScheduler.FromCurrentSynchronizationContext());
自定义任务调度器
除了使用.NET提供的调度器外,我们能够继承类TaskScheduler
来实现自己的任务调度器。这里不再展开,需要了解的可以参考Samples for Parallel Programming with the .NET Framework。
原文地址:https://www.cnblogs.com/chenbaoshun/p/10621819.html
.NET社区新闻,深度好文,欢迎访问公众号文章汇总 http://www.csharpkit.com