任务并行库和 SQL 连接

Question

我希望有人可以通过 TPL 和 SQL 连接确认这里实际发生了什么。

基本上，我有一个大型应用程序，它实质上是从 SQL Server 读取表，然后按顺序处理每一行。每一行的处理可能需要相当长的时间。所以，我想改变它以使用任务并行库，在数据表中的行中使用“Parallel.ForEach”。这似乎工作了一会儿（分钟），然后一切都变成了梨形......

“在从池中获取连接之前已经过了超时时间。这可能是因为所有池中的连接都在使用中并且达到了最大池大小。”

现在，我推测了以下（当然可能完全错误）。

“ForEach”为每一行创建任务，根据核心数（或其他）。让我们说 4 想要一个更好的主意。这四个任务中的每一个都有一行，然后开始处理它。 TPL 等到机器不太忙，然后再启动一些。我预计最多四个。

但这不是我观察到的——也不是我认为正在发生的事情。

所以...我写了一个快速测试（见下文）：

Sub Main()
    Dim tbl As New DataTable()

    FillTable(tbl)

    Parallel.ForEach(tbl.AsEnumerable(), AddressOf ProcessRow)

End Sub

Private n As Integer = 0

Sub ProcessRow(row As DataRow, state As ParallelLoopState)
    n += 1 ' I know... not thread safe
    Console.WriteLine("Starting thread {0}({1})", n, Thread.CurrentThread.ManagedThreadId)
    Using cnx As SqlConnection = New SqlConnection(My.Settings.ConnectionString)
        cnx.Open()
        Thread.Sleep(TimeSpan.FromMinutes(5))
        cnx.Close()
    End Using
    Console.WriteLine("Closing thread {0}({1})", n, Thread.CurrentThread.ManagedThreadId)
    n -= 1
End Sub

这比我对任务数量的猜测要多得多。所以，我推测 TPL 将任务启动到它认为会让我的机器忙碌的极限，但是嘿，这是什么，我们这里不是很忙，所以让我们开始更多。仍然不是很忙，所以......等等。（似乎每秒一个新任务 - 大致）。

这是合理的，但我希望它会在 30 秒后弹出 30 秒（SQL 连接超时），如果它获得 100 个打开的 SQL 连接 - 默认连接池大小 - 它不会不。

所以，为了缩小一点，我更改了我的连接字符串以限制最大池大小。

Sub Main()
    Dim tbl As New DataTable()

    Dim csb As New SqlConnectionStringBuilder(My.Settings.ConnectionString)
    csb.MaxPoolSize = 10
    csb.ApplicationName = "Test 1"
    My.Settings("ConnectionString") = csb.ToString()

    FillTable(tbl)

    Parallel.ForEach(tbl.AsEnumerable(), AddressOf ProcessRow)

End Sub

我计算了与 SQL 服务器的实际连接数，正如预期的那样，它是 10。但是我的应用程序已经启动了 26 个任务 - 然后挂起。因此，设置 SQL 的最大池大小以某种方式将任务数限制为 26，但为什么没有 27，尤其是为什么它不会因为池已满而在 11 时倒下？

显然，在某个地方，我要求做的工作比我的机器做的多，我可以在 ForEach 中添加“MaxDegreesOfParallelism”，但我对这里实际发生的事情很感兴趣。

PS。

实际上，在处理 26 个任务（我猜）5 分钟后，它确实 因原始 （达到最大池大小） 错误而崩溃。嗯？

谢谢。

编辑 1：

实际上，我现在认为任务中发生的情况（我的“ProcessRow”方法）是在 10 个成功的连接/任务之后，第 11 个 does 阻塞连接超时，然后确实得到原始异常 - 就像任何后续任务一样。

所以...我得出结论，TPL 正在以大约每秒 1 的速度创建任务，并且在任务 11 引发异常之前，它有足够的时间创建大约 26/27。然后所有后续任务也会抛出异常（大约相隔一秒），并且 TPL 停止创建新任务（因为它在一个或多个任务中获得未处理的异常？）

由于某种原因（尚未确定），ForEach 会挂起一段时间。如果我修改我的 ProcessRow 方法以使用状态说“停止”，它似乎没有效果。

Sub ProcessRow(row As DataRow, state As ParallelLoopState)
    n += 1
    Console.WriteLine("Starting thread {0}({1})", n, Thread.CurrentThread.ManagedThreadId)
    Try
        Using cnx As SqlConnection = fnNewConnection()
            Thread.Sleep(TimeSpan.FromMinutes(5))
        End Using
    Catch ex As Exception
        Console.WriteLine("Exception on thread {0}", Thread.CurrentThread.ManagedThreadId)
        state.Stop()
        Throw
    End Try
    Console.WriteLine("Closing thread {0}({1})", n, Thread.CurrentThread.ManagedThreadId)
    n -= 1
End Sub

编辑 2：

Dur... 长时间延迟的原因是，虽然任务 11 以后的任务都崩溃和烧毁，但任务 1 到 10 不会，并且都坐在那里睡了 5 分钟。 TPL 已停止创建新任务（因为它已创建的一个或多个任务中有未处理的异常），然后等待未崩溃的任务完成.

Answer 1

对原始问题的修改增加了更多细节，最终答案变得显而易见。

TPL 重复创建任务，因为它已创建的任务（基本上）是空闲的。这很好，直到连接池耗尽，此时需要新连接的任务等待一个可用，然后超时。与此同时，TPL 仍在创造更多的任务，所有这些都注定要失败。连接超时后，任务开始失败，随后的异常导致 TPL 停止创建新任务。然后，在引发 AggregateException 之前，TPL 会等待确实获得连接的任务完成。

Answer 2

TPL 不适用于受 IO 限制的工作。它具有启发式方法，用于控制处于活动状态的线程数。对于长时间运行和/或受 IO 限制的任务，这些启发式方法会失败，导致它在没有实际限制的情况下注入越来越多的线程。

使用 PLINQ 使用 WithDegreeOfParallelism 设置固定数量的线程。您可能应该测试不同的数量。 It could look like this.我已经在 SO 上写了很多关于这个主题的文章，但我现在找不到。

我不知道您为什么在示例中看到 恰好 26 个线程。请注意，当池耗尽时，建立连接的请求只会在超时后失败。整个系统是非常不确定的，我认为任何数量的线程都是合理的。