VB.NET，对线程池的并发结果进行排序的最佳实践？

Question

简而言之，我正在尝试对使用线程池的传入结果进行“排序”。我有一个功能性的解决方案，但世界上没有办法是最好的方法（它容易出现巨大的停顿）。所以我在这里！我将尝试指出正在发生的事情/需要发生的事情的要点，然后发布我当前的解决方案。

• 代码的目的是获取有关目录中文件的信息，然后将其写入文本文件。

• 我有一个列表 (Counter.ListOfFiles)，它是以特定方式排序的文件路径列表。这是指示我需要写入文本文件的顺序的指南。

• 我正在使用线程池来收集有关每个文件的信息，创建一个字符串构建器，其中包含准备写入文本文件的所有文本。然后我调用一个过程（SyncUpdate，包括在下面），从该线程发送 stringbuilder(strBld) 以及特定线程刚刚写给 stringbuilder about(Xref) 的文件的路径名称。

  李>

• 该过程包括一个同步锁，用于保持所有其他线程，直到找到传递正确信息的线程。当线程传递的外部参照与我的列表中的第一项（FirstListItem）匹配时，该“正确”信息。发生这种情况时，我会写入文本文件，删除列表中的第一项，然后使用下一个线程再次执行此操作。

我使用显示器的方式可能不是很好，事实上我毫不怀疑我正在以一种冒犯性的肆意方式使用它。基本上，当外部参照（来自线程）我列表中的第一项时，我正在为监视器做一个脉冲。我最初使用的是 monitor.wait，但它最终会放弃尝试对列表进行排序，即使在其他地方使用脉冲也是如此。我可能只是在笨拙地编码一些东西。不管怎样，我认为这不会改变任何事情。

基本上，问题归结为这样一个事实，即监视器将通过它在队列中的所有项目进行脉冲处理，而当我正在寻找的项目很有可能在队列中较早的某个地方传递给它时，或者其他什么现在它将再次对所有项目进行排序，然后再循环查找匹配的条件。这样做的结果是我的代码将遇到其中一个并需要大量时间才能完成。

我愿意相信我只是在工作中使用了错误的工具，或者只是没有正确使用我拥有的工具。我非常喜欢某种线程解决方案（不出所料，它要快得多！）。今天我一直在搞乱并行任务功能，很多东西看起来很有希望，但我对线程池的经验甚至更少，你可以看到我是如何滥用它的！也许有队列的东西？你明白了。我没有方向。任何人可以建议将不胜感激。谢谢！如果您需要任何其他信息，请告诉我。

  Private Sub SyncUpdateResource(strBld As Object, Xref As String)
    SyncLock (CType(strBld, StringBuilder))
        Dim FirstListitem As String = counter.ListOfFiles.First
        Do While Xref <> FirstListitem
            FirstListitem = Counter.ListOfFiles.First
    'This makes the code much faster for reasons I can only guess at.
            Thread.Sleep(5)
            Monitor.PulseAll(CType(strBld, StringBuilder))
        Loop
        Dim strVol As String = Form1.Volname
        Dim strLFPPath As String = Form1.txtPathDir
        My.Computer.FileSystem.WriteAllText(strLFPPath & "\" & strVol & ".txt", strBld.ToString, True)
        Counter.ListOfFiles.Remove(Xref)
    End SyncLock
End Sub

Answer 1

这是一个非常典型的多生产者、单消费者应用程序。唯一的问题是您必须在将结果写入输出之前对其进行排序。这并不难做到。因此，让我们暂时搁置该要求。

.NET 中实现生产者/消费者关系的最简单方法是使用BlockingCollection，这是一个线程安全的 FIFO 队列。基本上，你这样做：

在您的情况下，生产者线程获取项目，执行他们需要的任何处理，然后将项目放入队列中。不需要任何显式同步——BlockingCollection 类实现会为您完成。

您的消费者从队列中拉出东西并输出它们。你可以在我的文章Simple Multithreading, Part 2 中看到一个非常简单的例子。 （如果您只对代码感兴趣，请向下滚动到第三个示例。）该示例仅使用一个生产者和一个消费者，但如果您愿意，您可以拥有 N 个生产者。

您的要求有一点问题，即消费者不能在获取项目时将它们写入文件。它必须确保以正确的顺序编写它们。正如我所说，这并不难做到。

您需要的是某种优先级队列，如果项目乱序，您可以将其放置在该队列上。如果您希望出现乱序的项目数量不是很大，您的优先级队列可以是一个排序列表，甚至只是一个顺序列表。也就是说，如果您通常一次只有半打可能出现故障的项目，那么顺序列表就可以正常工作。

我会使用堆，因为它的性能很好。 .NET Framework 不提供堆，但我有一个简单的堆，它适用于这样的工作。见A Generic BinaryHeap Class。

这就是我编写消费者的方式（代码是伪 C#，但你可以很容易地转换它）。

这里假设您有一个名为sharedQueue 的BlockingCollection，其中包含这些项目。生产者将项目放在该队列上。消费者这样做：

var heap = new BinaryHeap<ItemType>();
foreach (var item in sharedQueue.GetConsumingEnumerable())
{
    if (item.SequenceKey == expectedSequenceKey)
    {
        // output this item
        // then check the heap to see if other items need to be output
        expectedSequenceKey = expectedSequenceKey + 1;
        while (heap.Count > 0 && heap.Peek().SequenceKey == expectedSequenceKey)
        {
            var heapItem = heap.RemoveRoot();
            // output heapItem
            expectedSequenceKey = expectedSequenceKey + 1;
        }
    }
    else
    {
        // item is out of order
        // put it on the heap
        heap.Insert(item);
    }
}
// if the heap contains items after everything is processed,
// then some error occurred.

这种方法的一个明显问题是，如果您的一个消费者崩溃或进入无限循环，堆可能会无限制地增长。但是，您的其他方法也可能会受到影响。如果您认为这是一个问题，则必须添加一些方法来跳过您认为永远不会出现的项目。或者杀死程序。什么的。

如果您没有二进制堆或不想使用二进制堆，您可以使用SortedList<ItemType> 做同样的事情。 SortedList 将比 List 快，但比 BinaryHeap 慢，如果列表中的项目数量相当大（几十个）。比这还少，可能是洗了。

我知道这是很多信息。我很乐意回答您的任何问题。