Parallel.ForEach并行处理列表时结果不匹配的原因及解决方法咨询
Parallel.ForEach并行处理列表时结果不匹配的原因及解决方法咨询
嗨,我来帮你理清楚这个并行处理的问题哈!
为什么会出现结果不匹配的情况?
核心问题出在你用的List<T>不是线程安全的集合,具体来说:
List<T>的内部实现没有同步机制:它的Add方法执行时,会先检查内部数组容量,不够就扩容,然后把元素放到当前索引的位置,再把Count加1。这些步骤都不是原子性的,多线程同时操作时,会互相干扰。- 竞争条件导致数据异常:比如两个线程同时执行
Add,可能会有一个线程的元素被覆盖,或者Count的数值没有正确更新,最后就会出现输出的集合元素数量和输入不匹配,甚至偶尔抛出异常的情况。
解决方法:给你三个可行的方案
方案1:使用线程安全集合(最推荐)
直接把List<Person>替换成.NET内置的线程安全集合,比如ConcurrentBag<T>,它专门为多线程并发添加元素设计,内部自带同步机制,不用自己处理锁:
List<string> list = new List<string>() { "John", "Mary", "Margaret", "Silvia", "Martha" }; // 替换成线程安全的ConcurrentBag ConcurrentBag<Person> people = new ConcurrentBag<Person>(); Console.WriteLine("There are {0} names on the list", list.Count.ToString()); Parallel.ForEach(list, name => { people.Add(new Person(name)); }); foreach (Person person in people) { Console.WriteLine("Name: {0,-10} Age: {1}", person.Name, person.Age.ToString()); } Console.WriteLine("There are {0} people on the list", people.Count.ToString());
这样修改后,不管多少线程并发,集合的元素数量都会和输入一致,也不会出现元素丢失的情况。
方案2:手动加锁控制并发
如果一定要用List<T>,可以给Add操作加锁,确保同一时间只有一个线程能修改集合:
List<string> list = new List<string>() { "John", "Mary", "Margaret", "Silvia", "Martha" }; List<Person> people = new List<Person>(); // 定义一个专属的锁对象 object _lockObj = new object(); Console.WriteLine("There are {0} names on the list", list.Count.ToString()); Parallel.ForEach(list, name => { var person = new Person(name); // 加锁后再执行添加操作 lock(_lockObj) { people.Add(person); } }); foreach (Person person in people) { Console.WriteLine("Name: {0,-10} Age: {1}", person.Name, person.Age.ToString()); } Console.WriteLine("There are {0} people on the list", people.Count.ToString());
不过这种方法会损失一部分并行性能,因为锁会让线程排队等待,适合对集合类型有特殊要求的场景。
方案3:使用Parallel.ForEach的本地状态重载(性能最优)
这个方法让每个线程先处理自己的本地集合,最后再合并到全局集合,减少锁的竞争,性能比直接加锁好很多:
List<string> list = new List<string>() { "John", "Mary", "Margaret", "Silvia", "Martha" }; List<Person> people = new List<Person>(); object _lockObj = new object(); Console.WriteLine("There are {0} names on the list", list.Count.ToString()); Parallel.ForEach( list, // 每个线程初始化自己的本地临时集合 () => new List<Person>(), // 每个元素处理后先添加到本地集合 (name, loopState, localList) => { localList.Add(new Person(name)); return localList; }, // 每个线程结束后,把本地集合批量合并到全局集合 localList => { lock(_lockObj) { people.AddRange(localList); } } ); foreach (Person person in people) { Console.WriteLine("Name: {0,-10} Age: {1}", person.Name, person.Age.ToString()); } Console.WriteLine("There are {0} people on the list", people.Count.ToString());
这种方式因为每个线程只在最后合并时加一次锁,而不是每次Add都加锁,并行效率更高,适合数据量比较大的场景。
额外说明
需要注意的是,Parallel.ForEach本身的执行顺序是不确定的,所以即使解决了集合的线程安全问题,输出的Person顺序也可能和输入的list顺序不一样——这是并行处理的正常现象,如果你需要保持顺序,可以使用Parallel.ForEach的可排序重载,或者后续自己对集合按Name排序。
现在你可以根据自己的需求选择对应的方案,最推荐的是方案1,简单高效,完全能解决你遇到的问题!




