c# - 并行的 AddOrUpdate 的 ConcurrentDictionary 和 ConcurrentBag

Question

使用 ConcurrentDictionary 和 ConcurrentBag 来添加或更新值是否正确。
基本上尝试如下，

拥有包含数百万条记录的文件并尝试处理和提取对象。

条目就像键值对，Key=WBAN 和 Value 作为对象。

 var cd = new ConcurrentDictionary<String, ConcurrentBag<Data>>();
 int count = 0;

 foreach (var line in File.ReadLines(path).AsParallel().WithDegreeOfParallelism(5))
 {
     var sInfo = line.Split(new char[] { ',' });
     cd.AddOrUpdate(sInfo[0], new ConcurrentBag<Data>(){ new Data()
     {
         WBAN =  sInfo[0],
             Date = string.IsNullOrEmpty(sInfo[1]) ? "" : sInfo[1],
             time = string.IsNullOrEmpty(sInfo[2]) ? "" : sInfo[2]
 }
     }
     ,
     (oldKey, oldValue) =>
     {
         oldValue.Add(new Data()
         {
             WBAN = sInfo[0],
             Date = string.IsNullOrEmpty(sInfo[1]) ? "" : sInfo[1],
             time = string.IsNullOrEmpty(sInfo[2]) ? "" : sInfo[2]
         });

         return oldValue;
     }
     );
 }

Answer 1

您的程序受 IO 限制，而不是受 CPU 限制，因此并行处理没有任何优势。

它是 IO 绑定(bind)的，因为您的程序在没有首先从文件中读取该行的情况下无法处理该行数据，而且通常来说，计算机从存储中读取数据的速度总是比它们处理它的速度慢得多。

由于您的程序在读取的每一行上只执行微不足道的字符串操作，我可以肯定地说，添加 Data 所需的时间为 99.9%。元素到 Dictionary<String,List<Data>>是计算机从文本文件中读取一行所需时间的一小部分。

另外，避免使用 File.ReadLines对于像这样的程序，因为这将首先将整个文件读入内存。

如果您查看我的解决方案，您会发现它使用了 StreamReader一个接一个地读取每一行，这意味着它不需要等到它首先将所有内容读入内存。

因此，要以最佳性能解析该文件，您不需要任何并发集合。
只是这个:

private static readonly Char[] _sep = new Char[] { ',' }; // Declared here to ensure only a single array allocation.

public static async Task< Dictionary<String,List<Data>> > ReadFileAsync( FileInfo file )
{
    const Int32 ONE_MEGABYTE = 1 * 1024 * 1024; // Use 1MB+ sized buffers for async IO. Not smaller buffers like 1024 or 4096 as those are for synchronous IO.

    Dictionary<String,List<Data>> dict = new Dictionary<String,List<Data>>( capacity: 1024 );


    using( FileStream fs = new FileStream( path, FileAccess.Read, FileMode.Open, FileShare.Read, ONE_MEGABYTE, FileOptions.Asynchronous | FileOptions.SequentialScan ) )
    using( StreamReader rdr = new StreamReader( fs ) )
    {
        String line;
        while( ( line = await rdr.ReadLineAsync().ConfigureAwait(false) ) != null )
        {
            String[] values = line.Split( sep );
            if( values.Length < 3 ) continue;

            Data d = new Data()
            {
                WBAN = values[0],
                Date = values[1],
                time = values[2]
            };

            if( !dict.TryGetValue( d.WBAN, out List<Data> list ) )
            {
                dict[ d.WBAN ] = list = new List<Data>();
            }

            list.Add( d );
        }
    }
}

更新:假设...
假设地说，因为文件 IO(尤其是异步 FileStream IO)使用大缓冲区(在本例中为 ONE_MEGABYTE 大小的缓冲区)，所以程序可以将每个缓冲区(按顺序读取)传递到并行处理器中。
然而问题在于，该缓冲区内的数据不能轻易地分配给各个线程:在这种情况下，因为一行的长度不是固定的，所以单个线程仍然需要读取整个缓冲区以找出缓冲区的位置。换行符是(从技术上讲，可以在某种程度上并行化，这将增加大量的复杂性(因为您还需要处理跨越缓冲区边界的行，或仅包含单行的缓冲区等)。
在这种小规模下，使用线程池和并发收集类型的开销将消除并行处理的加速，因为程序仍然很大程度上受 IO 限制。
现在，如果您有一个以千兆字节为单位的文件，使用 Data大小约为 1KB 的记录，然后我将详细说明如何做到这一点，因为在这种规模下，您可能会看到适度的性能提升。