asynchronous - 使用 Async.Parallel 进行调度

Question

有没有什么方法可以通过引入调度程序来限制/限制 Async.Parallel？我希望并行执行 Async<'a> 的 Seq，但不想超过特定的每小时限制。

我可以使用每个 Async<'a> 检查的共享可变变量，但我想尽可能避免这种情况。

Answer 1

在幕后，Async.Parallel 操作使用标准的 .NET 线程池。因此，您可以配置线程池，但这可能不是一个好主意(您不应该在线程池中阻塞线程)。

如果我想实现一些限制，我可能会为此创建一个 F# 代理。代理为您提供了一种非常简单的方式来协调并发性 - 它可能比使用可变变量(为此目的)使用更多代码，但它为您提供了一个很好的抽象:

// We can ask the agent to enqueue a new work item;
// and the agent sends itself a completed notification
type ThrottlingMessage = 
  | Enqueue of Async<unit>
  | Completed

let throttlingAgent limit = MailboxProcessor.Start(fun inbox -> async {
  // The agent body is not executing in parallel, 
  // so we can safely use mutable queue & counter 
  let queue = System.Collections.Generic.Queue<_>()
  let running = ref 0
  while true do
    // Enqueue new work items or decrement the counter
    // of how many tasks are running in the background
    let! msg = inbox.Receive()
    match msg with
    | Completed -> decr running
    | Enqueue w -> queue.Enqueue(w)
    // If we have less than limit & there is some work to
    // do, then start the work in the background!
    while running.Value < limit && queue.Count > 0 do
      let work = queue.Dequeue()
      incr running
      do! 
        // When the work completes, send 'Completed'
        // back to the agent to free a slot
        async { do! work
                inbox.Post(Completed) } 
        |> Async.StartChild
        |> Async.Ignore })

要使用它，您可以创建一个具有指定限制的代理，然后调用 Enqueue 来添加您的工作项:

let w = throttlingAgent 5 
for i in 0 .. 20 do 
  async { printfn "Starting %d" i
          do! Async.Sleep(1000)
          printfn "Done %d" i  }
  |> Enqueue
  |> w.Post

这解决的问题与您遇到的问题有点不同 - 但它应该显示方向(而不是有 Completed 通知，您可能想要一些 async 在后台每小时发送指定数量的“ token ”)。