F#中的异步及并行模式（1）：并行CPU及I/O计算
2010-03-03 23:37 by 老赵, 8450 visits最后还是忍不住翻译文章了。这系列的文章谈论的是F#中常见的异步及并行模式，作者为F#语言的主要设计者Don Syme。异步相关的编程是F#语言中最重要的优势之一（我甚至在考虑“之一”两个字能否去掉）。F#是一门非常有特色的语言，是一门能够开阔眼界，改变您编程思路的语言，它经过了几年设计以及多个预览之后终于要正式露面了——此刻不上，更待何时。
介绍
F#是一门并行（parallel）及响应式（reactive）语言。这个说法意味着一个F#程序可以存在多个进行中的运算（如使用.NET线程进行F#计算），或是多个等待中的回应（如等待事件或消息的回调函数及代理对象）。
F#的异步表达式是简化异步及响应式程序编写的方式之一。在这篇及今后的文章中，我会探讨一些使用F#进行异步编程的基本方式──大致说来，它们都是F#异步编程时使用的模式。这里我假设您已经掌握了async的基本使用方式，如入门指南中的内容。
我们从两个简单的设计模式开始：CPU异步并行（Parallel CPU Asyncs）和I/O异步并行（Paralle I/O Asyncs）。
模式1：CPU异步并行
首先来了解第一个模式：CPU异步并行，这意味着并行地开展一系列的CPU密集型计算。下面的代码计算的是斐波那契数列，它会将这些计算进行并行地调配：
let rec fib x = if x <= 2 then 1 else fib(x-1) + fib(x-2) let fibs = Async.Parallel [ for i in 0..40 -> async { return fib(i) } ] |> Async.RunSynchronously
结果是：
val fibs : int array =
[|1; 1; 2; 3; 5; 8; 13; 21; 34; 55; 89; 144; 233; 377; 610; 987; 1597; 2584;
4181; 6765; 10946; 17711; 28657; 46368; 75025; 121393; 196418; 317811;
514229; 832040; 1346269; 2178309; 3524578; 5702887; 9227465; 14930352;
24157817; 39088169; 63245986; 102334155|]
上面的代码展示了并行CPU异步计算模式的要素：
- “async { … }”用于指定一系列的CPU任务。
- 这些任务使用Async.Parallel进行fork-join式的组合。
在这里，我们使用Async.RunSynchronously方法来执行组合后的任务，这会启动一个异步任务，并同步地等待其最后结果。您可以使用这个模式来完成各种CPU并行（例如对矩阵乘法进行划分和并行计算）或是批量处理任务。
模式2：I/O异步并行
现在我们已经展示了在F#中进行CPU密集型并行编程的方式。F#异步编程的重点之一，便是可以用相同的方式进行CPU和I/O密集型的计算。这便是我们的第二种模式：I/O异步并行，即同时开展多个I/O操作（也被称为overlapped I/O）。例如下面的代码便并行地请求多个Web页面，并响应每个请求的回复，再返回收集到的结果。
open System open System.Net open Microsoft.FSharp.Control.WebExtensions let http url = async { let req = WebRequest.Create(Uri url) use! resp = req.AsyncGetResponse() use stream = resp.GetResponseStream() use reader = new StreamReader(stream) let contents = reader.ReadToEnd() return contents } let sites = ["http://www.bing.com"; "http://www.google.com"; "http://www.yahoo.com"; "http://www.search.com"] let htmlOfSites = Async.Parallel [for site in sites -> http site ] |> Async.RunSynchronously
上面的代码示例展示了I/O异步并行模式的基础：
- “async { … }”用于编写任务，其中包含了一些异步I/O。
- 这些任务使用Async.Parallel进行fork-join式的组合。
在这里，我们使用Async.RunSynchronously方法来执行组合后的任务，这会启动一个异步任务，并同步地等待其最后结果。
使用let!（或与它类似的资源释放指令use!）是进行异步操作的基础方法。例如：
let! resp = req.AsyncGetResponse()
上面这行代码会“响应”一个HTTP GET操作所得到的回复，即async { … }在AsyncGetResponse操作完成之后的部分。然而，在等待响应的过程中并不会阻塞任何.NET或操作系统的线程：只有活动的CPU密集型运算会使用下层的.NET或操作系统线程。与此不同，等待中的响应操作（例如回调函数，事件处理程序和代理对象）资源占用非常少，几乎只相当于一个注册好的对象而已。因此，您可以同时拥有数千个甚至数百万个等待中的响应操作。例如，一个典型的GUI应用程序会注册一些事件处理程序，而一个典型Web爬虫会为每个发出的请求注册一个回调函数。
在上面的代码中，我们使用了“use!”而不是“let!”，这表示Web请求相关的资源会在变量超出字面的作用域之后得到释放。
I/O并行的美妙之处在于其伸缩性。在多核的环境下，如果您可以充分利用计算资源，则通常会获得2倍、4倍甚至8倍的性能提高。而在I/O并行编程中，您可以同时进行成百上千的I/O操作（不过实际的并行效果还要取决于您的操作系统和网络连接状况），这意味着10倍、100倍、1000倍甚至更多的性能增强──而这一切在一台单核的机器上也可以实现。例如，这里有一个使用F#异步功能的示例，而最终它们可以在一个IronPython应用程序中使用。
许多现代应用程序都是I/O密集型应用，因此这些设计模式在实践中都有很重要的意义。
始于GUI线程，终于GUI线程
这两个设计模式有个重要的变化，这便是使用Async.StartWithContinuations来代替Async.RunSynchronously方法。在一个并行操作开启之后，您可以指定三个函数，分别在它成功、失败或取消时调用。
对于诸如“我想要获得一个异步操作的结果，但我不能使用RunSynchronously方法”之类的问题，您便应该考虑：
- 使用let!（或use!）把这个异步操作作为更大的异步任务的一部分，或者
- 使用Async.StartWithContinuations方法执行异步操作
在那些需要在GUI线程上发起异步操作的场景中，Async.StartWithContinuations方法尤其有用。因为，您不会因此阻塞住GUI线程，而且可以在异步操作完成后直接进行GUI的更新。例如，在F# JAOO Tutorial的BingTranslator示例中便使用了这个做法──您可以在本文结尾浏览它的完整代码，不过这里最值得关注的部分则是在点击“Translate”按钮之后发生的事情：
button.Click.Add(fun args -> let text = textBox.Text translated.Text <- "Translating..." let task = async { let! languages = httpLines languageUri let! fromLang = detectLanguage text let! results = Async.Parallel [for lang in languages -> translateText (text, fromLang, lang)] return (fromLang,results) } Async.StartWithContinuations( task, (fun (fromLang,results) -> for (toLang, translatedText) in results do translated.Text <- translated.Text + sprintf "\r\n%s --> %s: \"%s\"" fromLang toLang translatedText), (fun exn -> MessageBox.Show(sprintf "An error occurred: %A" exn) |> ignore), (fun cxn -> MessageBox.Show(sprintf "A cancellation error ocurred: %A" cxn) |> ignore)))
高亮的部分，尤其是在async块里的部分，展示了使用Async.Parallel将一种语言并行地翻译成多种语言的做法。这个异步组合操作由Async.StartWithContinuations发起，它会在遇到第一个I/O操作时立即返回（译注：存疑，为什么是在遇上I/O操作才返回？），并指定了三个函数，分别在异步操作的成功，失败或取消时调用。以下是任务完成后的截图（不过在此不保证翻译的准确性……）：
Async.StartWithContinuations有一个重要的特性：如果异步操作由GUI线程发起（例如一个SynchronizationContext.Current不为null的线程），那么操作完成后的回调函数也是在GUI线程中调用的。这使GUI更新操作变的十分安全。F#异步类库允许您组合多个I/O任务，并在GUI线程中直接使用，而无需您亲自从后台线程中更新GUI元素。在以后的文章中我们会进行更详细地解释。
关于Async.Parallel工作方式：
- 在执行时，由Async.Parallel组合而成的异步操作会通过一个等待计算的队列来逐步发起。与大部分进行异步处理的类库一样，它在内部使用的是QueueUserWorkItem方法。当然，我们也有办法使用分离的队列，在以后的文章中我们会进行一些讨论。
- Async.Parallel方法并没有什么神奇之处，您也完全可以使用Microsoft.FSharp.Control.Async类库中的其他原语来定义您自己的异步组合方式──例如Async.StartChild方法。我们会在以后的文章中讨论这个话题。
更多示例
在F# JAOO Tutorial包含多个使用这些模式的示例代码：
- BingTranslator.fsx与BingTranslatorShort.fsx：使用F#调用REST API，它们与其他基于Web的HTTP服务的调用方式十分类似。文末包含了示例的完整代码。
- AsyncImages.fsx：并行磁盘I/O及图像处理。
- PeriodicTable.fsx：调用一个Web服务，并行地获取原子质量。
本文模式的限制
上文介绍的两个并行模式有一些限制。很明显，使用Async.Parallel生成的异步操作在执行时十分“安静”──比方说，它们无法返回进度或部分的结果。为此，我们需要构建一个更为“丰富”的对象，它会在部分操作完成之后触发一些事件。在以后的文章中我们会来关注这样的设计模式。
此外，Async.Parallel只能处理固定数量的任务。在以后的文章中，我们会遇到很多一边处理一边生成任务的情况。换个方式来看，即Async.Parallel无法处理即时获得的消息──例如，除了取消任务之外，一个代理对象的工作进度是可以得到控制的。
总结
CPU异步并行与I/O异步并行，是F#异步编程中最为简单的两种设计模式，而简单的事物往往也是非常重要而强大的。请注意，两种模式的不同之处，仅仅在于I/O并行使用了包含了I/O请求的async块，以及一些额外的CPU任务，如创建请求对象及后续处理。
在今后的文章里，我们会关注F#中其他一些并行及响应式编程方面的设计方式，包括：
- 从GUI线程中发起异步操作
- 定义轻量级异步代理对象
- 使用async定义后台工作程序
- 使用async构建.NET任务
- 使用async调用.NET的APM模式
- 取消异步操作
BingTranslator代码示例
以下是BingTranslator的示例代码，在运行时您需要申请一个Live API 1.1 AppID。请注意，这个示例需要根据Bing API 2.0进行适当调整，至少在2.0中已经不包含这里的语言检测API了──不过这些代码仍然是不错的示例：
点此展开
原文：Async and Parallel Design Patterns in F#: Parallelizing CPU and I/O Computations
好吧，我想做F#在国内的推广者。