asynchronous - 如何限制传递给 Async.Parallel 的大量异步工作流

Question

我有一个包含大量小型异步数据库查询的数组；例如:

// I actually have a more complex function that
// accepts name/value pairs for query parameters.
let runSql connString sql = async {
    use connection = new SqlConnection(connString)
    use command = new SqlCommand(sql, connection)
    do! connection.OpenAsync() |> Async.AwaitIAsyncResult |> Async.Ignore
    return! command.ExecuteScalarAsync() |> Async.AwaitTask
    }

let getName (id:Guid) = async {
    // I actually use a parameterized query
    let querySql = "SELECT Name FROM Entities WHERE ID = '" + id.ToString() + "'"
    return! runSql connectionString querySql
    }

let ids : Guid array = getSixtyThousandIds()

let asyncWorkflows = ids |> Array.map getName
//...

现在，问题是:下一个表达式同时运行所有 60K 工作流，从而淹没服务器。这导致许多 SqlCommand 超时；由于我不理解并且(不需要理解它们)尚未调查的原因，它通常还会导致客户端(这是 F# 交互的)内存不足异常:

//...
let names =
    asyncWorkflows
    |> Async.Parallel
    |> Async.RunSynchronously

我编写了一个粗略的函数来批处理请求:

let batch batchSize asyncs = async {
    let batches = asyncs
                  |> Seq.mapi (fun i a -> i, a)
                  |> Seq.groupBy (fst >> fun n -> n / batchSize)
                  |> Seq.map (snd >> Seq.map snd)
                  |> Seq.map Async.Parallel
    let results = ref []
    for batch in batches do
        let! result = batch
        results := (result :: !results)
    return (!results |> List.rev |> Seq.collect id |> Array.ofSeq)
}

要使用此函数，我将 Async.Parallel 替换为 batch 20(或其他整数值):

let names =
    asyncWorkflows
    |> batch 20
    |> Async.RunSynchronously

这工作得相当好，但我更喜欢有一个系统在一个完成后立即启动每个新的异步，所以我总是在等待 N active SqlCommand s(当然，直到我结束)。

问题:

我是在重新发明轮子吗？换句话说，是否已经有库函数可以做到这一点？ (以某种方式研究利用 ParallelEnumerable.WithDegreeOfParallelism 是否有利可图？)

如果不是，我应该如何实现连续队列而不是一系列离散批次？

我不是主要寻求改进现有代码的建议，但仍然会带着兴趣和感激地接受这些建议。

Answer 1

FSharpx.Control offers an Async.ParallelWithThrottle function .我不确定它是否是最好的实现，如 it uses SemaphoreSlim .但是易用性很棒，而且由于我的应用程序不需要顶级性能，因此对我来说效果很好。虽然因为它是一个库，如果有人知道如何使它变得更好，那么让库开箱即用总是一件好事，这样我们其他人就可以使用有效的代码并完成我们的工作!