F#与ASP.NET（2）：使用F#实现基于事件的异步模式
2010-04-05 20:37 by 老赵, 2951 visits在上一篇文章中，我们的简单讨论了.NET中两种异步模型以及它们在异常处理上的区别，并且简单观察了ASP.NET MVC 2中异步Action的编写方式。从中我们得知，ASP.NET MVC 2的异步Action并非使用了传统基于Begin/End的异步编程模型，而是另一种基于事件的异步模式。此外，ASP.NET MVC 2对于这种异步模式提供了必要的支持，使此方面的程序设计变得相对简单一些。但是，简单的原因主要还是在于已经由其他组件提供了良好的，基于事件的异步模式。那么现在我们就来看看一般我们应该如何来实现这样的功能，以及F#是如何美化我们的生活的吧。
异步数据传输
我们为什么要异步，主要目的之一还是提高I/O操作的伸缩性。I/O操作主要是I/O设备的事情，这些设备在好好工作的时候，是不需要系统花心思进行照看的，它们只要能够在完成指定工作后通知系统就可以了。这也是异步I/O高效的原理，因为它将程序从等待I/O完成的苦闷中解脱出来，这对用户或是系统来说都是一件绝好的事情。
那么说到I/O操作，最典型的场景之一便是数据传输了。比如有两个数据流streamIn和streamOut，我们需要异步地从streamIn中读取数据，并异步地写入到streamOut中。在这个过程中，我们使用一个相对较小的byte数组作为缓存空间，这样程序在进行数据传输时便不会占用太多内存。那么，如果现在需要您编写一个组件完成这样的数据传输工作，并使用标准的基于事件的异步模式释放出来，您会怎么做？
再啰嗦一次，基于事件的异步模式，要求在任务完成时使用事件进行提示。同时在出错的时候将异常对象保存在事件的参数中。现在我已经帮您写好了这样的事件参数：
public class CompletedEventArgs : EventArgs { public CompletedEventArgs(Exception ex) { this.Error = ex; } public Exception Error { get; private set; } }
那么接下来的工作就交给您了，加油！
嗯？那么快就写完啦？再想想？如果真确定了，就展开下面的代码对比一下吧：
展开代码
隐藏代码 public class AsyncTransfer { private Stream m_streamIn; private Stream m_streamOut; public AsyncTransfer(Stream streamIn, Stream streamOut) { this.m_streamIn = streamIn; this.m_streamOut = streamOut; } public void StartAsync() { byte[] buffer = new byte[1024]; this.m_streamIn.BeginRead( buffer, 0, buffer.Length, this.EndReadInputStreamCallback, buffer); } private void EndReadInputStreamCallback(IAsyncResult ar) { var buffer = (byte[])ar.AsyncState; int lengthRead; try { lengthRead = this.m_streamIn.EndRead(ar); } catch (Exception ex) { this.OnCompleted(ex); return; } if (lengthRead <= 0) { this.OnCompleted(null); } else { try { this.m_streamOut.BeginWrite( buffer, 0, lengthRead, this.EndWriteOutputStreamCallback, buffer); } catch (Exception ex) { this.OnCompleted(ex); } } } private void EndWriteOutputStreamCallback(IAsyncResult ar) { try { this.m_streamOut.EndWrite(ar); var buffer = (byte[])ar.AsyncState; this.m_streamIn.BeginRead( buffer, 0, buffer.Length, this.EndReadInputStreamCallback, buffer); } catch (Exception ex) { this.OnCompleted(ex); } } private void OnCompleted(Exception ex) { var handler = this.Completed; if (handler != null) { handler(this, new CompletedEventArgs(ex)); } } public event EventHandler<CompletedEventArgs> Completed; }
是不是很复杂的样子？
编写异步程序，基本则意味着要将原本同步的调用拆成两段：发起及回调，这样便让上下文状态的保存便的困难起来。幸运的是，C#这门语言提供了方便好用的匿名函数语法，这对于编写一个回调函数来说已经非常容易了。但是，如果需要真正写一个稳定、安全的异步程序，需要做的事情还有很多。例如，一次异步操作结束之后会执行一个回调函数，那么如果在这个回调函数中抛出了一个异常那该怎么办？如果不正确处理这个异常，轻则造成资源泄露，重则造成进程退出。因此在每个回调函数中，您会发现try...catch块是必不可少的——甚至还需要两段。
更复杂的可能还是在于逻辑控制上。这样一个数据传输操作很显然需要循环——读一段，写一段。但是由于需要编写成二段式的异步调用，因此程序的逻辑会被拆得七零八落，我们没法使用一个while块包围整段逻辑。
编写一个异步程序本来就是那么复杂。
编写简单的代理
嗯，我们继续。现在我们已经有了一个异步传输数据的组件，就用它来做一些有趣的事情吧。例如，我们可以在ASP.NET应用程序中建立一个简单的代理，即给定一个URL，在服务器端发起这样一个请求，并将这个URL的数据传输到客户端来。简单起见，除了进行数据传输之外，我们只需要简单地输出Content Type头信息即可。
写好了吗？我也写了一个，仅供参考：
展开代码
隐藏代码 public class AsyncWebTransfer { private WebRequest m_request; private WebResponse m_response; private HttpContextBase m_context; private string m_url; public AsyncWebTransfer(HttpContextBase context, string url) { this.m_context = context; this.m_url = url; } public void StartAsync() { this.m_request = WebRequest.Create(this.m_url); this.m_request.BeginGetResponse(this.EndGetResponseCallback, null); } private void EndGetResponseCallback(IAsyncResult ar) { try { this.m_response = this.m_request.EndGetResponse(ar); this.m_context.Response.ContentType = this.m_response.ContentType; var streamIn = this.m_response.GetResponseStream(); var streamOut = this.m_context.Response.OutputStream; var transfer = new AsyncTransfer(streamIn, streamOut); transfer.Completed += (sender, args) => this.OnCompleted(args.Error); transfer.StartAsync(); } catch(Exception ex) { this.OnCompleted(ex); } } private void OnCompleted(Exception ex) { if (this.m_response != null) { this.m_response.Close(); this.m_response = null; } var handler = this.Completed; if (handler != null) { handler(this, new CompletedEventArgs(ex)); } } public event EventHandler<CompletedEventArgs> Completed; }
如果说之前的AsyncTransfer类是基于“Begin/End异步编程模型”实现的基于事件的异步模式，那么AsyncWebTransfer便是基于“基于事件的异步模式”实现的基于事件的异步模式了。嗯，似乎有点绕口，不过我相信这段代码对您来说还是不难理解的。
使用F#完成异步工作
好吧，我承认，前面的代码我也觉得很复杂，很难写。事实上我已经很久没有写过这样的代码了，咎其原因还是被F#给宠坏了。尝试了C# 2.0之后我便抛弃了Java语言，熟悉了C# 3.0之后我用C# 2.0就快写不了程序了，而使用了F#进行异步编程之后，我就再也没有使用C#写过异步操作了。
那么我们就来看看F#是如何进行异步数据传输的吧：
let rec transferAsync (streamIn: Stream) (streamOut: Stream) buffer = async { let! lengthRead = streamIn.AsyncRead(buffer, 0, buffer.Length) if lengthRead > 0 then do! streamOut.AsyncWrite(buffer, 0, lengthRead) do! transferAsync streamIn streamOut buffer }
喔喔，上面的代码利用了尾递归进行不断地数据传输，我们也可以使用传统的while循环来实现这个功能（关于ref的作用可参考这篇文章）：
let transferImperativelyAsync (streamIn: Stream) (streamOut: Stream) buffer = async { let hasData = ref true while (hasData.Value) do let! lengthRead = streamIn.AsyncRead(buffer, 0, buffer.Length) if lengthRead > 0 then do! streamOut.AsyncWrite(buffer, 0, lengthRead) else hasData := false }
有了transferAsync函数，编写一个资源请求的代理也是几分钟的事情：
let webTransferAsync (context: HttpContextBase) (url: string) = async { let request = WebRequest.Create(url) use! response = request.GetResponseAsync() context.Response.ContentType <- response.ContentType let streamIn = response.GetResponseStream() let streamOut = context.Response.OutputStream let buffer = Array.zeroCreate 1024 do! transferAsync streamIn streamOut buffer }
没错，就是这么简单，这就是F#中编写异步任务方式（关于这方面的更多细节，您可以阅读F#的主要设计者Don Syme所撰写的一系列文章）。在执行这两个函数时（当然确切地说，是执行这两个函数所生成的异步工作流），便会在出现“感叹号”的操作之处自动分成二段式的异步调用，但是在程序的写法上和同步代码可谓毫无二致。
使用F#实现基于事件的异步模式
当然，光有上面的代码还不够，因为这样的代码无法交给C#代码来使用，我们还需要将它们封装成基于事件的异步模式。不过这也非常简单，使用一个通用的抽象基类即可：
[<AbstractClass>] type AsyncWorker(asyncWork: Async) = let completed = new Event () [<CLIEvent>] member e.Completed = completed.Publish member e.StartAsync() = Async.StartWithContinuations (asyncWork, (fun _ -> completed.Trigger(new CompletedEventArgs(null))), (fun ex -> completed.Trigger(new CompletedEventArgs(ex))), (fun ex -> ex |> ignore))
在使用F#进行面向对象开发时，由于不需要C#的架子代码，它实现相同的结构一般都会显得紧凑不少（不过在我看来，C#在进行一般的命令式编程时还是比F#来的方便一些）。在StartAsync方法中，我们使用Async.StartWithContinuations发起一个异步工作流，而这个异步工作流便是从构造函数中传入的具体任务。StartWithContinuations方法的后三个参数分别是成功时的回调，失败后的回调，以及任务取消后的回调（我们并不需要这个回调，随意提供一个即可）。
您可能会说，难道F#中不需要异常处理，不需要资源释放吗？当然需要。只不过：1) 异常处理已经由StartWithContinuations统一完成了，我们只要按照“同步式”代码的写法编写逻辑，也就是说，从语义上说您可以看作存在一个巨大的try...catch围绕着整段代码。2) 而对于资源释放来说，您可以发现在webTransferAsync方法中有一个use!指令，这便是告诉F#的异步框架，在整个异步工作流结束之后需要调用这个资源的Dispose方法——没错，您可以把它看作是一种能在异步环境下工作的C# using关键字。
有了AsyncWorker类之后，AsyncTransfer和WebAsyncTransfer类也可轻易实现了：
type AsyncTransfer(streamIn: Stream, streamOut: Stream) = inherit AsyncWorker( Transfer.transferAsync streamIn streamOut (Array.zeroCreate 1024)) type AsyncWebTransfer(context: HttpContextBase, url: string) = inherit AsyncWorker(Transfer.webTransferAsync context url)
最后，只要在ASP.NET MVC中使用即可：
public void LoadFsAsync(string url) { AsyncManager.OutstandingOperations.Increment(); var transfer = new FSharpAsync.AsyncWebTransfer(HttpContext, url); transfer.Completed += (sender, args) => AsyncManager.OutstandingOperations.Decrement(); transfer.StartAsync(); } public ActionResult LoadFsCompleted() { return new EmptyResult(); }
事实上，在ImageController中我还提供了一个LoadAsync及对应的LoadCompleted方法，它们使用的是利用C#实现的AsyncWebTransfer类。猜猜看这样的代码长成什么样？其实只是将上面的AsyncWebTransfer的命名空间改成CSharpAsync而已——F#与其它.NET代码是真正做到无缝集成的。
总结
这便是F#的伟大之处。时常有朋友会问我为什么对F#有那么大的兴趣，我想，如果借助F#可以用十分之一的时间，十分之一的代码行数，写出执行效果相同，但可维护性高出好几倍的程序来——我为什么会不感兴趣呢？
您可以在这里访问到本文的示例代码，其中我在WebApp项目中实现了简单的入口页面，您访问“/image”便会出现两个文本框，您填入一个URL（例如某幅图片）并提交，它会将URL提交至“/image/load”或“/image/loadfs”中，它们分别使用了C#和F#实现的AsyncWebTransfer类，从效果上说两者完全相同。
相关文章
- F#与ASP.NET（1）：基于事件的异步模式与异步Action
- F#与ASP.NET（2）：使用F#实现基于事件的异步模式
老赵不容易啊，同步打理两博客~~下载看下实例......