任务并行库和SQL连接

Question

我希望有人能证实TPL和SQL连接实际发生了什么。

基本上，我有一个大型应用程序，它从SQL Server中读取一个表，然后按顺序处理每一行。每行的处理可能需要相当长的时间。因此，我想将其更改为使用任务并行库，在数据表中的行中使用“Parallel.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会将任务激发到它认为会让我的机器忙碌的极限，但嘿，这是什么，我们这里不是很忙，所以让我们开始更多。仍然不是很忙，所以…等等（看起来像是一秒钟一个新任务-大致如此）。

这是合理的，但我预计它会在获得100个打开的SQL连接（默认连接池大小）后弹出30秒（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 server的实际连接数，如预期的那样，是10个。但我的应用程序启动了26个任务，然后挂起。因此，为SQL设置最大池大小以某种方式将任务数限制为26个，但为什么没有27个，尤其是，为什么它不降为11个，因为池已满？

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

附言

实际上，在（我猜）5分钟内完成26项任务后，它确实出现了最初的（达到最大池大小）错误。呵呵？

谢谢

编辑1：

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

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

由于某种原因（尚未确定），Foreach than挂起了一段时间。如果我修改我的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不是为IO绑定的工作而设计的。它有启发式方法来引导活动线程的数量。这些启发式方法对于长时间运行和/或IO绑定的任务失败，导致它注入越来越多的线程而没有实际限制。

使用PLINQ使用并行度设置固定数量的线程。你可能应该测试不同的量。它可能看起来像这样。关于这个话题，我已经写了很多，但我现在找不到。

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

Answer 2

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

TPL重复创建任务，因为它创建的任务（基本上）处于空闲状态。这在连接池耗尽之前没有问题，此时需要新连接的任务等待连接可用，然后超时。与此同时，第三方物流仍在创造更多注定要失败的任务。连接超时后，任务开始失败，随后的异常会导致TPL停止创建新任务。然后，TPL等待确实获得连接的任务完成，然后抛出AggregateException。