f# - 在 F# 中一次强制执行一个异步 Observable

Question

我有一个可观察的事物序列需要映射到一个 C# 任务。这些 C# 任务不应同时运行，而应一个接一个地运行其他。基本上，我需要实现与 C# 等效的 F#问题:

Enforcing one async observable at a time

天真地翻译这段 C# 代码会得到如下内容:

let run (idx:int) (delay:int) =
    async {
        sprintf "start: %i (%i)" idx delay |> System.Diagnostics.Trace.WriteLine
        let! t = System.Threading.Tasks.Task.Delay(delay) |> Async.AwaitTask
        sprintf "finish: %i" idx  |> System.Diagnostics.Trace.WriteLine
        t
    }

    Observable.generate (new Random()) (fun _ -> true) id (fun s -> s.Next(250, 500))
    |> Observable.take 20
    |> Observable.mapi(fun idx delay -> idx, delay)
    |> Observable.bind(fun (idx, delay) -> Observable.ofAsync (run idx delay))
    |> Observable.subscribe ignore
    |> ignore

这不能按预期工作，因为我不会在任何地方等待结果。有没有甚至有没有一种方法可以在 F# 中执行此操作而不阻塞线程，就像 C# 的 wait 那样？

Answer 1

F# 中有一个方便的库，称为 AsyncSeq: https://www.nuget.org/packages/FSharp.Control.AsyncSeq/

与 IAsyncEnumerable<T> 非常相似添加到 C# 8.0 中，这为您提供了一个很好的解决方案。

解决方案:

open System
open FSharp.Control
open FSharp.Control.Reactive

[<EntryPoint>]
let main _ =

    let run (idx:int) (delay:int) =
        async {
            sprintf "start: %i (%i)" idx delay |> Console.WriteLine
            do! Async.Sleep delay
            sprintf "finish: %i" idx  |> Console.WriteLine
        }

    Observable.generate (new Random()) (fun _ -> true) id (fun s -> s.Next(250, 500))
    |> Observable.take 20
    |> Observable.mapi(fun idx delay -> idx, delay)
    |> AsyncSeq.ofObservableBuffered
    |> AsyncSeq.iterAsync (fun (idx,delay) -> run idx delay)
    |> Async.RunSynchronously

    0

AsyncSeq.ofObservableBuffered订阅您的 Observable 的工作并充当 AsyncSeq您可以在其上进行管道传输的源。最后我们调用Async.RunSynchronously将其启动并等待入口点线程。

注意:我还更新了run当它返回时Async<Async<unit>>我认为这不是有意的。