F#中的"链接"异步函数

Question

我在F#中创建了一个函数来从Yahoo恢复历史数据(F#的经典异步示例):

let getCSV ticker dStart dEnd =
async   {
        let query = getFileUrl ticker dStart dEnd
        let req = WebRequest.Create(query)
        use! resp = req.AsyncGetResponse()
        use stream= resp.GetResponseStream()
        use reader = new StreamReader(stream)
        let content = reader.ReadToEnd()
        let ts = parseData content
        return ts
        }

现在,我可以通过执行以下操作异步运行此函数:

let test=
    ["MSFT";"YHOO"]
    |>List.map (fun x -> getCSV x (DateTime.Parse("01.01.2000")) (DateTime.Parse("01.01.2010")))
    |> Async.Parallel
    |> Async.RunSynchronously

好,这很酷.

现在,我想知道的是如何应用一些功能,这是价格的历史:

例如:

let getReturns (prices:(DateTime *float)list) =
    [for i in 1..(prices.Length-1) -> i]
    |> List.map (fun i ->(fst (List.nth prices i), (snd (List.nth prices i))/(snd (List.nth prices (i-1) )) - 1.0))

所以这样做的琐碎方法是:

let test2=
    ["MSFT";"YHOO"]
    |>List.map (fun x -> getCSV x (DateTime.Parse("01.01.2000")) (DateTime.Parse("01.01.2010")))
    |> Async.Parallel
    |> Async.RunSynchronously
    |> Array.map getReturns;;

但是,getReturns每下载和解析一个文件就会执行该函数.

我想知道的是,如果可以在下载仍在进行时开始执行第二个功能:一旦完成MSFT,就不需要等到YHOO完成计算它的返回...

我知道我可以修改,getCSV但我想知道是否有办法"链接" getReturn功能而无需更改以前编写的模块...

Answer 1

我通常会直接在异步工作流中编写对函数的调用.这主要是风格或偏好的问题 - 我认为使用异步工作流编写的代码通常更明确,并且不经常使用高阶函数(尽管它们有时仍然有用):

let test=
    [ for stock in ["MSFT";"YHOO"] ->
        async { let! data = getCSV stock (DateTime(2000, 1, 1)) (DateTime(2010, 1, 1))
                return getReturns data } ]
    |> Async.Parallel
    |> Async.RunSynchronously 

这意味着并行执行的工作流首先获取数据,然后调用getRteurns以提取数据.然后将整个操作并行化.

或者,您可以使用Joel的解决方案(修改getReturns函数以便它采用异步工作流并返回异步工作流)或定义一个函数Async.map,该函数采用异步工作流并构造一个将某些函数应用于结果的新函数.

使用原始getReturns功能,您可以编写:

let test=
    ["MSFT";"YHOO"]
    // For every stock name, generate an asynchronous workflow
    |> List.map (fun x -> getCSV x (DateTime(2000, 1, 1)) (DateTime(2010, 1, 1)))
    // For every workflow, transform it into a workflow that 
    // applies 'getReturns' to the result of the original workflow
    |> List.map (Async.map getReturns)
    // Run them all in parallel
    |> Async.Parallel
    |> Async.RunSynchronously

定义Async.map很简单:

module Async =
  let map f workflow = async {
    let! res = workflow
    return f res }