我需要帮助创建一个 TaskScheduler 以防止线程过载

Question

我想将工作人员添加到队列中，但只有前 N 个工作人员并行处理。我找到的所有示例都在 C# 中。

这对程序员来说可能很简单，但我不是。我对 VB 的了解足以编写简单的程序。

但是我的第一个应用程序运行良好，直到它突然达到 100% CPU 然后崩溃。请帮忙（是的，我已经浪费了 5 个小时的工作时间搜索才发布此内容......）

更多上下文：对具有超过 100 万个目录/子目录的文件服务器执行目录结构、文件和权限的递归清单。

流程连续运行，但需要数月才能完成。管理层已经在我的脖子上呼吸了。当我尝试使用 Tasks 时，它会进入大约 1000 个线程，然后达到 100% CPU，停止响应，然后崩溃。这是在具有 112 GB RAM 的 16 核服务器上。

--添加 因此，通过使用信号量提供的示例，这就是我放入的内容：

Public Class InvDir
    Private mSm as Semaphore
    Public Sub New(ByVal maxPrc As Integer)
        mSm = New Semaphore(maxPrc, maxPrc)
    End Sub

    Public Sub GetInventory(ByVal Path As String, ByRef Totals As Object, ByRef MyData As Object)
        mSm.WaitOne()

        Task.Factory.StartNew(Sub()
                Dim CurDir As New IO.DirectoryInfo(Path)
                Totals.SubDirectoryCount += CurDir.GetDirectories().Count
                Totals.FilesCount += CurDir.GetFiles().Count
                For Each CurFile As IO.FileInfo in CurDir.EnumerateFiles()
                    MyData.AddFile(CurFile.FileName, CurFile.Extension, CurFile.FullName, CurFile.Length)
                Next
                End Sub).ContinueWith(Function(x) mSm.Release())
    End Sub
End Class

Answer 1

您正在尝试使用磁盘 I/O 进行多线程处理。它可能会变得越来越慢因为你正在向它抛出更多线程。无论有多少线程，磁盘在物理上一次只能查找一个位置。 （事实上​​，你提到它是串行工作的。）

如果您确实想限制并发线程的数量，您可以使用Semaphore。信号量类似于同步锁，只是您可以指定一次允许多少个线程执行代码。在下面的示例中，信号量允许三个线程执行。除此之外，必须等到一个完成。来自 MSDN 页面的一些修改代码：

Public Class Example

    ' A semaphore that simulates a limited resource pool.
    '
    Private Shared _pool As Semaphore

    <MTAThread> _
    Public Shared Sub Main()
        ' Create a semaphore that can satisfy up to three
        ' concurrent requests. Use an initial count of zero,
        ' so that the entire semaphore count is initially
        ' owned by the main program thread.
        '
        _pool = New Semaphore(0, 3)          

    End Sub

    Private Sub SomeWorkerMethod()
        'This is the method that would be called using a Task.
        _pool.WaitOne()
        Try
            'Do whatever
        Finally
            _pool.Release()
        End Try
    End Sub
End Class

每个新线程都必须调用_pool.WaitOne()。这告诉它等待轮到它，直到执行的线程少于三个。每个线程都会阻塞，直到信号量允许它通过。

每个线程还必须调用_pool.Release() 让信号量知道它可以允许下一个等待线程开始。这很重要，即使有例外。如果线程不调用Release()，那么信号量将永远阻塞它们。

如果真的需要五个月的时间，那么克隆驱动器并对同一驱动器的多个实例运行检查，每个实例查看不同的部分呢？