6

我正在使用代理（MailboxProcessor）在需要响应的地方进行一些有状态的处理。

  • 来电者使用MailboxProcessor.PostAndAsyncReply
  • 在代理内部，会给出一个响应AsyncReplyChannel.Reply

但是，通过查看 f# 源代码，我发现在响应传递之前，代理不会处理下一条消息。总的来说，这是一件好事。但在我的情况下，代理更希望继续处理消息而不是等待响应传递。

做这样的事情来传递响应是否有问题？（或者有更好的选择吗？）

async { replyChannel.Reply response } |> Async.Start

我意识到这种方法并不能保证响应将按顺序传递。我没关系。

参考示例

// agent code
let doWork data =
    async { ... ; return response }

let rec loop ( inbox : MailboxProcessor<_> ) =
    async {
        let! msg = inbox.Receive()
        match msg with
        | None ->
            return ()

        | Some ( data, replyChannel ) ->
            let! response = doWork data
            replyChannel.Reply response (* waits for delivery, vs below *)
            // async { replyChannel.Reply response } |> Async.Start
            return! loop inbox
    }

let agent =
    MailboxProcessor.Start(loop)

// caller code
async {
    let! response =
        agent.PostAndAsyncReply(fun replyChannel -> Some (data, replyChannel))
    ...
}
4

1 回答 1

0

FSharp.Control.AsyncSeq在邮箱处理器之上放置了一个更友好的面孔。异步序列更容易遵循，但是默认实现映射并行具有与描述相同的问题，等待序列中的前一个元素被映射以保留顺序。

所以我创建了一个新函数，它只是原始的 AsyncSeq.mapAsyncParallel，经过修改使其不再是真正的映射，因为它是无序的，但它确实映射了所有内容，并且惰性 seq 会随着工作的完成而进展。

AsyncSeq.mapAsyncParallelUnordered 的完整源代码

let mapAsyncParallelUnordered (f:'a -> Async<'b>) (s:AsyncSeq<'a>) : AsyncSeq<'b> = asyncSeq {
  use mb = MailboxProcessor.Start (fun _ -> async.Return())
  let! err =
    s 
    |> AsyncSeq.iterAsyncParallel (fun a -> async {
      let! b = f a
      mb.Post (Some b) })
    |> Async.map (fun _ -> mb.Post None)
    |> Async.StartChildAsTask
  yield! 
    AsyncSeq.replicateUntilNoneAsync (Task.chooseTask (err |> Task.taskFault) (async.Delay mb.Receive))
  }

下面是一个示例，说明我如何在使用 SSLlabs 免费且非常慢的 api 很容易过载的工具中使用它。parallelProcessHost返回AsyncSeq由 webapi 请求生成的延迟，因此AsyncSeq.mapAsyncParallelUnordered AsyncSeq.toListAsync实际运行请求并允许控制台在输入时打印输出结果，与发送的顺序无关。

完整来源

let! es = 
    hosts
    |> Seq.indexed
    |> AsyncSeq.ofSeq
    |> AsyncSeq.map parallelProcessHost
    |> AsyncSeq.mapAsyncParallelUnordered AsyncSeq.toListAsync
    |> AsyncSeq.indexed
    |> AsyncSeq.map (fun (i, tail) -> (consoleN "-- %d of %i --- %O --" (i+1L) totalHosts (DateTime.UtcNow - startTime)) :: tail )
    |> AsyncSeq.collect AsyncSeq.ofSeq
    |> AsyncSeq.map stdoutOrStatus //Write out to console
    |> AsyncSeq.fold (|||) ErrorStatus.Okay
于 2020-02-17T15:37:54.837 回答