15.6.2Task使用 组合异步操作

Posted 2021-01-28 kikyoqiang

　　对于C# 5异步特性，我最喜欢的一点是它可以自然而然地组合在一起。这表现为两种不同的 方式。最明显的是，异步方法返回任务，并通常会调用其他返回任务的方法。这些方法可以是直 接的异步操作（如链的最底部），也可以是更多的异步方法。所有的包装和拆包都需要将结果转 换为任务，反向操作则由编译器完成。

　　另一种组合形式是，创建与操作无关的构建块来管理任务的处理。这些构建块无须知道任务 在做什么，而只是单纯待在 Task<T> 的抽象级别。这有点像LINQ操作符，只是面向的是任务而 不是序列。框架中内置了一些构建块，但也可以自行创建。

1. 在单个调用中收集结果

　　例如，尝试获取若干URL。15.3.6节中一次性获取了所有URL，并在完成任务后立即停止获 取。假设这次要启动多个并行请求，然后每得到一个URL就记录下结果。记住，异步方法返回的 是已经运行的任务，因此可非常轻松地为每个URL启动一个任务：

1             string[] urls = new string[] { "http://stackoverflow.com", "http://www.google.com", "http://csharpindepth.com" };
2             var tasks = urls.Select(async url =>
3             {
4                 using (var client = new HttpClient())
5                 {
6                     return await client.GetStringAsync(url);
7                 }
8             }).ToList();

　　注意，需调用 ToList() 来具体化LINQ查询。这保证了每个任务将只启动一次。否则每次迭 代 tasks 时，将会再次获取字符串。（如不释放 HttpClient ，代码会更加简单，但即便如此，代 码也不是很难看。）

　　TPL提供了一个 Task.WhenAll 方法，从而将各有一个结果的多个任务组合成一个包含多个 结果的任务。常用的方法重载签名如下所示：

        public static Task<TResult[]> WhenAll<TResult>(IEnumerable<Task<TResult>> tasks);

　　这个声明看上去非常糟糕，但在真正使用时，会发现其方法目的非常单纯。你将得到一个 List<Task<string>> ，因此可以写为：

            string[] results = await Task.WhenAll(tasks);

　　所有任务均已结束，并将结果收集到一个数组中后，等待方可终止。本章前面讲过，如果多 个任务抛出异常，则只有第一个异常会立即抛出，但可总是迭代这些任务，以找到具体失败的任 务及其失败原因，或使用代码清单15-2中所示的 WithAggregatedException 扩展方法。

　　如果只关注第一个返回的请求，则可使用 Task.WhenAny 方法。该方法不会等待第一个成功 完成的任务，而只会等待第一个到达终点状态的任务。

　　本例中，你可能想要点特别的做法。在任务完成后报告全部结果可能会更有用些。

2. 在全部完成时收集结果

　　Task.WhenAll 是.NET内置的转换构建块（transformational building block），接下来将介绍如何以类似的方式构建自己的方法。TAP文档中含有类似的示例代码，从而创建了 Interleaved 方法，这里将介绍另一版本。

　　代码清单15-12旨在传递一个输入任务的序列，并返回一个输出任务的序列。两个序列中任 务的结果是相同的，但存在一个重要差异，即输出任务的完成顺序与输入完全一样，因此可以一 次 await 一个任务，并可立即得到任务结果。这听上去有些神奇，对我来说也是如此，因此我们 来看看代码，研究一下它的工作原理。

 1         public static IEnumerable<Task<T>> InCompletionOrder<T>(this IEnumerable<Task<T>> source)
 2         {
 3             var inputs = source.ToList();
 4             var boxes = inputs.Select(x => new TaskCompletionSource<T>()).ToList();
 5 
 6             int currentIndex = -1;
 7             foreach (var task in inputs)
 8             {
 9                 task.ContinueWith(completed =>
10                 {
11                     var nextBox = boxes[Interlocked.Increment(ref currentIndex)];
12                     PropagateResult(completed, nextBox);
13                 }, TaskContinuationOptions.ExecuteSynchronously);
14             }
15             return boxes.Select(box => box.Task);
16         }

　　代码清单15-12依赖TPL中一个非常重要的类型，即 TaskCompletionSource<T> 。该类型 可用于创建一个尚未含有结果的 Task ，并在之后提供结果（或异常）。它和 AsyncTaskMethod Builder<T> 都建立在相同的基础结构之上。后者为异步方法提供返回的 Task ，并在方法体完成 时，将带结果的任务向外传播。

　　为什么会用这么奇怪的变量名（ boxes ）呢？我常常把任务想象成纸箱，这些纸箱承诺 （promise）在某个时刻，其内部会含有值或错误。 TaskCompletionSource<T> 就像是背面有洞 的箱子，你可以把它给别人，然后再偷偷地把值从洞口塞进去 ① 。这正是 PropagateResult 方法 的作用，不过它没那么有意思，所以此处不予列出，基本上它会将已完成的 Task<T> 的结果传播 到 TaskCompletionSource<T> 中。如果原始任务正常完成，则将返回值复制到 Task CompletionSource<T> 中。如果原始任务产生了错误，则可将异常复制到 TaskCompletion Source<T> 中。取消原始任务后， TaskCompletionSource<T> 也会随之被取消。

　　真正聪明的部分是（我对此说法不承担任何责任——有人发邮件建议我加入这一免责声明）， 在该方法运行时，它并不知道哪个 TaskCompletionSource<T> 会对应哪个输入任务，而只是将 相同的后续操作附加到各任务上，然后由后续操作来寻找下一个 TaskCompletionSource<T> （通过对一个计数器进行原子地累加）并传播结果。也就是说，它会按照原始任务的输出顺序对箱子进行填充。

　　图15-5展示了三个输入任务，以及相应的由方法返回的输出任务。即使输入任务的顺序与方 法返回的顺序不同，输出任务的顺序也会与之相同。 有了这个绝妙的扩展方法后，即可编写代码清单15-13，从而得到一组URL，并行地对每个 URL发起请求，并在请求完成时写下各页面的长度，然后返回总长度。

 1         static void Main()
 2         {
 3             var task = ShowPageLengthsAsync("http://stackoverflow.com", "http://www.google.com", "http://csharpindepth.com");
 4             Console.WriteLine("Total length: {0}", task.Result);
 5         }
 6         static async Task<int> ShowPageLengthsAsync(params string[] urls)
 7         {
 8             var tasks = urls.Select(async url =>
 9             { 
10                 using (var client = new HttpClient())
11                 {
12                     return await client.GetStringAsync(url);
13                 }
14             }).ToList();
15 
16             int total = 0;
17             foreach (var task in tasks.InCompletionOrder())
18             {
19                 string page = await task;
20                 Console.WriteLine("Got page length {0}", page.Length);
21                 total += page.Length;
22             }
23             return total;
24         }

　　代码清单15-13存在两个小问题。
　　1.一个任务失败，则整个异步操作都将失败，并且不会保留结果。这也许没问题，但也可能希望能够将每次失败记录下来。（与.NET 4不同，不处理任务异常，则默认不会让进程当掉，但至少应考虑对其他任务产生的影响。）
　   2.失去对页面转向具体URL的跟踪。
这两个问题都可以通过少量代码轻松解决，也可以进一步提取成可复用的构建块。举这些例子并不是为了满足个别需求，而是为了让你接受组合带来的各种可能性。TAP白皮书中并不是只有 Interleaved 这一个例子，它还包括很多概念，并附带一些有助于理解的示例。