1

我有一个关于使用 BlockingCollection 和 Dictionary 编写代码的问题。

我的目标是读取一堆文本文件并以并行方式处理它们。处理后的数据将存储在 BlockingCollection 实例中，以便可以将这些处理后的数据写入文件。

我想使用 BlockingCollection 的原因是......

(1) 在 GenerateDataFiles() 做 CPU 密集型工作时节省时间，同时消费者 Task 可以做 IO 相关工作，并且

(2) 与我在将所有已处理数据写入文件之前将所有已处理数据存储在列表中的情况相比，减少内存使用量。

对于（2），如果我在将所有数据写入文件之前存储它们，内存消耗超过我的桌面可以承受的（因为它读取超过 30GB 的数据），所以我必须使用这种生产者 - 消费者的方法。

此外，我在 BlockingCollection 实例（或字典）中插入键值对时遇到问题。请指出进行数据插入的正确方法。

以下代码是我尝试解决问题的方法。因为我是 BlockingCollection 的新手，所以我可能在这方面犯了一些错误。请提出一些更改（和修改的代码），以便我解决问题。

class SampleClass
{
    static void Main(string[] args)
    {            
        SampleClass sampleClass = new SampleClass();
        sampleClass.run();
    }

    private void run()
    {
        Task consumer = Task.Factory.StartNew(() => WriteDataToFiles());
        GenerateDataFiles();
    }

    BlockingCollection<Dictionary<string, List<string>>> bc = new BlockingCollection<Dictionary<string, List<string>>>();

    private void GenerateDataFiles()
    {
        DirectoryInfo directory = new DirectoryInfo(@"D:\Data\");
        FileInfo[] array_FileInfo = directory.GetFiles("*.txt", SearchOption.TopDirectoryOnly);

        Parallel.ForEach(array_FileInfo, fileInfo => 
        {
            string[] array_Lines = File.ReadAllLines(fileInfo.FullName);

            // do some CPU-intensive data parsing and then add the processed data to the blocking collection
            // It has to be inserted in pairs (key = file path, value = list of strings to be written to this file)

        });
    }

    private void WriteDataToFiles()
    {
        foreach (var item in bc.GetConsumingEnumerable())
        {
            foreach (var key in item.Keys)
            {
                File.WriteAllLines(key, item[key]);
            }
        }

    }
}
4

1 回答 1

1

考虑在. Tuple_ 此外，您需要调用来结束in 。DictionaryBlockingCollectionCompleteAdding()foreachWriteDataToFiles

BlockingCollection<Tuple<string, List<string>>> bc = new BlockingCollection<Tuple<string, List<string>>>();

private void GenerateDataFiles()
{
    DirectoryInfo directory = new DirectoryInfo(@"D:\Data\");
    FileInfo[] array_FileInfo = directory.GetFiles("*.txt", SearchOption.TopDirectoryOnly);

    Parallel.ForEach(array_FileInfo, fileInfo => 
    {
        string[] array_Lines = File.ReadAllLines(fileInfo.FullName);

        // do some CPU-intensive data parsing and then add the processed data to the blocking collection
        // It has to be inserted in pairs (key = file path, value = list of strings to be written to this file)
        List<string> processedData = new List<string>();  // ... and add content
        bc.Add(new Tuple<string, List<string>>(fileInfo.FullName, processedData));
    });
    bc.CompleteAdding();
}

private void WriteDataToFiles()
{
    foreach (var tuple in bc.GetConsumingEnumerable())
    {
        File.WriteAllLines(tuple.Item1, tuple.Item2);
    }
}
于 2014-12-28T20:19:11.640 回答