使用 Async.Parallel 进行调度 [英] Scheduling with Async.Parallel

问题描述

有什么办法可以通过引入调度程序来限制/限制 Async.Parallel 吗?我希望并行执行 Async<'a> 的 Seq，但不想超过某个小时限制.

Is there any way that Async.Parallel can be limited/ throttled by introducing a scheduler? I'm looking to execute a Seq of Async<'a> in parallel but don't want to exceed a certain hourly-limit.

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

I could use a shared mutable variable that each Async<'a> examines but I'd like to avoid this if possible.

推荐答案

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

Under the cover, the Async.Parallel operation uses the standard .NET thread pool. So, you could configure the thread pool, but that's probably not a good idea (you should not be blocking threads in a thread pool).

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

If I wanted to implement some throttling, I would probably create an F# agent for this. Agents give you a pretty simple way to coordinate the concurrency - it is probably more code than using mutable variable (for this purpose), but it gives you a nice abstraction:

// 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 以添加您的工作项:

To use this, you can create an agent with a specified limit and then call Enqueue to add your work items:

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> 在后台每小时发送指定数量的令牌").

This is solving a bit different problem than the one you have - but it should show the direction (rather than having the Completed notification, you probably want to have some async in the background that sends a specified number of "tokens" every hour).

这篇关于使用 Async.Parallel 进行调度的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！