13.1.4 创建基元工作流

13.1.4 创建基元工作流



F# PowerPack 库包含许多重要的 I/O 操作异步版本，但是，它可不能包括所有。为此，F# 库还提供了构建自己的基元工作流的方法。如果要运行在工作流内部的操作，使用的标准 .NET 模式，并提供了 BeginOperation 和 EndOperation 方法，可以使用 Async.FromBeginEnd 方法。如果你把这两种方法给它作为参数值，它会返回一个异步工作流。

可以无需阻塞线程而执行的其他操作是可用的。例如，我们可能要等待一个特定的事件发生，当它触发时，继续执行这个工作流。清单 13.4 创建了一个基元，使用计时器，等待指定的毫秒数，然后继续工作流。



Listing 13.4 Implementing asynchronous waiting (F# Interacitve)



> module MyAsync =
let Sleep(time) =
Async.FromContinuations(fun (cont, econt, ccont) –>
let tmr = new System.Timers.Timer(time, AutoReset = false)
tmr.Elapsed.Add(fun _ -> cont())
tmr.Start()
);;
(...)

> Async.RunSynchronously(async {
printfn "Starting..."
do! MyAsync.Sleep(1000.0)
printfn "Finished!"
});;
Starting...
Finished!
val it : unit = ()



相同的功能，已经在 F# 库中可用了，所以，它并不只是一个玩具的例子。它由 Async.Sleep 实现，我们将在本章后面需要使用它。当然，我们可能阻塞该工作流，使用同步版本，Thread.Sleep，但有一个重要的区别。这个方法会阻塞线程，而我们的函数创建一个计时器，并返回该线程到 .NET 线程池。这意味着，当我们在使用自己的基元，.NET 运行时可以并行执行工作流，而不受任何限制。

Sleep 函数取我们想要延迟处理的毫秒数，使用 Async.FromContinuations 方法来构建工作流。此方法相当密切反映了工作流的内部结构。该参数值是 lambda 函数，将在工作流启动时执行。Lambda 取三个连续的元组作为参数值。第一个函数应在操作成功完成时调用，第二个应在操作引发异常时调用。类似于 Async<'T> 类型的声明，来自早前的侧边栏，第三次连续，可以触发取消工作流。在 lambda 体中，我们创建一个计时器，指定 Elapsed 事件的处理程序。这个处理程序只运行成功连续。

清单 13.4 显示，创建了自己的基元后，用简单的代码段来使用它。因为它返回 unit 值，我们将用 do! 基元，而不是 let!。当这个代码执行时，它构造了带这个处理程序的计时器，然后启动它。当指定的时间流逝，系统从线程池取一个可用的线程，运行事件处理程序，依次执行计算的其余部分（在本例中，打印到屏幕）。



在 C# 中的异步工作流



已经有无数次尝试，简化 C# 中的异步编程，但是，没有可用的库，像异步工作流语法一样整齐地工作。从最终用户的角度来看（只将代码打包在 async 块中），F# 语法是非常简单的，这在 C# 中是很难实现的。

我们已经看到，LINQ 查询大概相当于 F# 的计算表达式，所以，你可能会忍不住要实现 Select 和 SelectMany 操作。原则上，写异步操作使用查询表达式，它是有可能的，但是，我们要在查询内部使用的语法是有限制的。有趣的是，C# 迭代器也可以用于此目的。这种做法在"在 C# 中使用迭代器的异步编程"文章中有叙述（在 http://tomasp.net/blog/csharpasync.aspx）。大多数使用这种技术的、最真实的库是 Jeffrey Richter 的 PowerThreading 库 [Richter，2009年]。

基于 C# 迭代最复杂的一个库是并发性和协调运行库 (CCR) [Chrysanthakopoulos 和 Singh，2005年]。这个库被开发作为微软机器人工作室（Microsoft Robotics studio）的一部分，响应能力和异步处理，对任何应用程序都是至关重要的。可以在 Jeffery Richter 的文章"并发事务"中找到有关这个库的详细信息 [Richter，2006年]。



是时候开始为更实用目的而使用异步工作流了。在下一节，我们将会看到由世界银行所提供的数据服务，讨论如何使用异步工作流来调用。