选择第一个异步结果

Question

是否有异步操作符来获取两个异步值首先返回的值（Async<_>）？

例如，给定两个 Async<_> 值，其中一个 A1 在 1 秒后返回，A2 在 2 秒后返回，那么我想要 A1 的结果。

原因是我想为异步序列实现一个交错函数，这样如果有两个异步序列像这样“定义”（空格表示时间，就像大理石图一样）：

S1 = -+-----+------------+----+
S2 = ---+-------+----------+-----+

然后我想生成一个新的异步序列，行为如下：

S3 = -+-+---+---+--------+-+--+--+

交错 S1 S2 = S3

但是两个这样做，我可能需要一种异步选择运算符来选择选择值。

我认为这就像 Go 中的“选择”，您可以从两个通道中获取第一个可用值。

TPL 有一个名为 Task.WhenAny 的函数——我可能在这里需要类似的东西。

Answer 1

我认为该运算符在 F# 库中不可用。要结合现有操作，您可以使用Async.StartAsTask，然后使用现有的Task.WhenAny 运算符。但是，我不确定在取消方面会有什么表现。

另一个选项是使用在F# Snippets web site 上实现的Async.Choose 运算符。这不是特别优雅，但它应该可以解决问题！为了使答案独立，代码附在下面。

/// Creates an asynchronous workflow that non-deterministically returns the 
/// result of one of the two specified workflows (the one that completes
/// first). This is similar to Task.WaitAny.
static member Choose(a, b) : Async<'T> = 
    Async.FromContinuations(fun (cont, econt, ccont) ->
      // Results from the two 
      let result1 = ref (Choice1Of3())
      let result2 = ref (Choice1Of3())
      let handled = ref false
      let lockObj = new obj()
      let synchronized f = lock lockObj f

      // Called when one of the workflows completes
      let complete () = 
        let op =
          synchronized (fun () ->
            // If we already handled result (and called continuation)
            // then ignore. Otherwise, if the computation succeeds, then
            // run the continuation and mark state as handled.
            // Only throw if both workflows failed.
            match !handled, !result1, !result2 with 
            | true, _, _ -> ignore
            | false, (Choice2Of3 value), _ 
            | false, _, (Choice2Of3 value) -> 
                handled := true
                (fun () -> cont value)
            | false, Choice3Of3 e1, Choice3Of3 e2 -> 
                handled := true; 
                (fun () -> 
                    econt (new AggregateException
                                ("Both clauses of a choice failed.", [| e1; e2 |])))
            | false, Choice1Of3 _, Choice3Of3 _ 
            | false, Choice3Of3 _, Choice1Of3 _ 
            | false, Choice1Of3 _, Choice1Of3 _ -> ignore )
        op() 

      // Run a workflow and write result (or exception to a ref cell
      let run resCell workflow = async {
        try
          let! res = workflow
          synchronized (fun () -> resCell := Choice2Of3 res)
        with e ->
          synchronized (fun () -> resCell := Choice3Of3 e)
        complete() }

      // Start both work items in thread pool
      Async.Start(run result1 a)
      Async.Start(run result2 b) )

Answer 2

托马斯已经回答了确切的问题。但是，您可能有兴趣知道我的 F# 的 Hopac 库直接支持 Concurrent ML -style 一等高阶选择性事件，称为 alternatives，它直接提供 choose -combinator 并提供比 Go 的 select 语句更具表现力的并发抽象机制。

关于交错两个异步序列的更具体问题，我最近开始尝试如何使用 Hopac 完成 Rx 风格的编程。我想出的一种可能的方法是定义一种短暂的事件流。你可以在这里找到实验代码：

• Alts.fsi
• Alts.fs

如您所见，为事件流定义的操作之一是merge。您正在寻找的内容可能在语义上略有不同，但使用 Hopac 风格的替代方案（或并发 ML 风格的事件）可能很容易实现。