使用任务并行库答案

【问题标题】：Using the Task Parallel Library使用任务并行库
【发布时间】：2015-01-14 17:04:09
【问题描述】：

我刚开始在 .NET 4.5 中编写和使用 TPL，想知道您是否可以帮助我。

基本上，我的 WPF 程序对一组可能在 5 到 1000 万之间的数据库记录执行繁重的工作。我想利用 TPL 库来有效地管理这项工作，同时保持 UI 响应。

我的场景如下，我计划使用 3 个单独的任务来检索每个 300 万的数据，并将它们提供给我的“工人”。

我想要一个单独的任务，从上述任务的所有项目中获取每个项目并对其进行一些处理。

所以我想要一个所有 3 个任务都可以放入项目的项目的全局列表，我的第四个任务，然后应该从这个主列表中一次选择一个项目并对其进行一些处理，如果列表是，它应该等待为空以便再次填充等。

您能否提供一些关于如何使用 TPL 执行此操作的指导？我是 TPL 的新手。一些示例代码会很好。

【问题讨论】：

这听起来很复杂。在来自数据库的结果流上运行的 Parallel.ForEach 循环怎么样？
这取决于您需要处理的工作类型。对于 CPU 密集型计算，Parallel.ForEach() 可能是正确的选择。对于 I/O 绑定（例如文件和数据库连接），等待异步方法的操作会更高效。

标签： c#

【解决方案1】：

考虑将每个Task 中的项目存储在一个全局的线程安全列表对象中，例如ConcurrentStack 或ConcurrentQueue。

【讨论】：

【解决方案2】：

如果我对问题描述的理解正确，您有一个生产者/消费者场景，其中有三个生产者，每个生产者都从数据库中检索独立的记录，还有一个消费者。

如果是这种情况，那么在我看来，您希望以BlockingCollection<T> 开头，其中T 是生产者从数据库中检索的对象类型。要使用的基础集合取决于您要如何处理数据。最常见的场景是先进先出和无序，分别用ConcurrentQueue<T>和ConcurrentBag<T>表示。

这是一个简单的代码示例，演示了它是如何工作的：

BlockingCollection<int> consumeFrom = new BlockingCollection<int>();
int producerCount = 3;

for (int i = 0; i < producerCount; i++)
{
    int taskValue = i;

    // Make dummy task for example
    Task.Run(() =>
    {
        for (int j = 0; j < 10; j++)
        {
            Thread.Sleep(1000);
            consumeFrom.Add(taskValue * 10 + j);
        }

        if (Interlocked.Decrement(ref producerCount) == 0)
        {
            consumeFrom.CompleteAdding();
        }
    });
}

foreach (int i in consumeFrom.GetConsumingEnumerable())
{
    Console.WriteLine(i);
}

请注意，这使用默认的基础数据源ConcurrentQueue<T>。您可以将IProducerConsumerCollection<T> 的其他实现提供给BlockingCollection<T> 构造函数，例如：

BlockingCollection<int> consumeFrom =
    new BlockingCollection<int>(new ConcurrentBag<int>());

【讨论】：