在Java `parallelStream`中使用嵌套循环时，因共享可变状态导致数据不一致是一个常见问题。本文深入探讨了数据竞争的原理，并提供了两种主要解决方案：一是利用`CopyOnWriteArrayList`等并发集合确保线程安全，二是采用`collect()`操作进行无状态或局部状态的聚合。同时，文章还对比了`forEach()`与`forEachOrdered()`的性能差异，旨在帮助开发者构建高效且结果准确的并行处理逻辑。
引言：Java并行流的优势与潜在陷阱
Java 8引入的Stream API极大地简化了集合操作，而parallelStream()则进一步提供了利用多核处理器进行并行处理的能力，从而在数据量较大时显著提升性能。然而，并行处理并非没有代价。当多个线程同时访问和修改同一个共享资源（即共享可变状态）时，如果不采取适当的同步措施，就可能导致数据竞争（Race Condition），进而产生错误或不一致的结果。这正是许多开发者在使用parallelStream并发现结果不符合预期时遇到的核心问题。
问题剖析：嵌套循环与数据竞争
在给定的场景中，开发者使用parallelStream().forEach()来迭代一个Person列表，并在每个Person的lambda表达式内部执行嵌套循环，以根据personId从另一个列表中获取相关数据并创建“组合”，最终将这些组合添加到一个结果集中。
import java.util.ArrayList;import java.util.List;// 示例：Person, Address, PersonAddressPair 类定义class Person { private Integer id; private String name; public Person(Integer id, String name) { this.id = id; this.name = name; } public Integer getId() { return id; } public String getName() { return name; }}class Address { private Integer personId; private String city; public Address(Integer personId, String city) { this.personId = personId; this.city = city; } public Integer getPersonId() { return personId; } public String getCity() { return city; }}class PersonAddressPair { Person person; Address address; public PersonAddressPair(Person person, Address address) { this.person = person; this.address = address; } // ... toString(), equals(), hashCode()}public class DataProcessor { // 模拟数据获取 private List getPersonDetails() { List persons = new ArrayList(); for (int i = 0; i < 1000; i++) persons.add(new Person(i, "Person " + i)); return persons; } private List getAddressDetails() { List addresses = new ArrayList(); for (int i = 0; i < 1000; i++) { if (i % 2 == 0) addresses.add(new Address(i, "City " + i)); } return addresses; } public List processDataIncorrectly() { List personList = getPersonDetails(); List addressList = getAddressDetails(); // !! 这是一个非线程安全的List，在并行环境下写入会导致数据丢失或不一致 !! List resultSet = new ArrayList(); personList.parallelStream().forEach(person -> { // 嵌套循环，根据personId匹配地址并生成组合 for (Address address : addressList) { if (address.getPersonId().equals(person.getId())) { // 多个线程可能同时尝试修改 resultSet，导致数据竞争 // 例如：一个线程正在添加元素，另一个线程同时读取或修改其内部结构， // 结果可能丢失部分添加操作，或导致内部状态损坏。 resultSet.add(new PersonAddressPair(person, address)); } } }); return resultSet; }}
ArrayList、HashSet等标准的Java集合类都不是线程安全的。这意味着它们的设计并未考虑多个线程同时对其进行修改的情况。当多个线程并发地调用add()方法时，可能会发生以下问题：
闪念贝壳
闪念贝壳是一款AI 驱动的智能语音笔记，随时随地用语音记录你的每一个想法。
218 查看详情 
立即学习“Java免费学习笔记（深入）”；
丢失更新（Lost Update）：一个线程的修改可能被另一个线程的修改覆盖，导致部分数据未能正确写入。数据不一致（Inconsistent State）：集合的内部结构（如数组大小、元素数量等）可能在并发修改下变得不一致，导致后续操作（如size()、迭代）返回错误的结果或抛出异常。
这就是为什么每次执行都会得到不同且不正确结果（例如，预期6k记录，实际2k且每次变化）的原因。
解决方案一：采用并发集合
最直接的解决方案是使用Java并发包（java.util.concurrent）中提供的线程安全集合。这些集合专门设计用于在多线程环境下安全地操作。对于本场景，如果主要的瓶颈在于向共享列表添加元素，CopyOnWriteArrayList或CopyOnWriteArraySet是合适的选择。
CopyOnWriteArrayList的特点：
写时复制（Copy-On-Write）：当列表被修改（添加、删除、设置）时，它会创建一个内部数组的新副本，并在新副本上执行修改。读操作则始终在旧副本上进行，因此读操作是完全无锁的，非常高效。线程安全：所有修改操作都通过锁来保证原子性，确保了数据的一致性。适用场景：非常适合读操作远多于写操作的场景。如果写操作非常频繁，由于每次修改都会复制底层数组，其性能可能会低于其他并发集合。
import java.util.List;import java.util.concurrent.CopyOnWriteArrayList;// ... Person, Address, PersonAddressPair 类定义// ... getPersonDetails(), getAddressDetails() 方法public class DataProcessor { // ... (其他方法) ... public List processDataWithConcurrentCollection() { List personList = getPersonDetails(); List addressList = getAddressDetails(); // 使用CopyOnWriteArrayList，它在写入时会创建内部数组的副本，确保线程安全 List resultSet = new CopyOnWriteArrayList(); personList.parallelStream().forEach(person -> { for (Address address : addressList) { if (address.
以上就是Java并行流中嵌套循环导致结果不一致的问题：原因与解决方案的详细内容，更多请关注创想鸟其它相关文章！
版权声明：本文内容由互联网用户自发贡献，该文观点仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。
如发现本站有涉嫌抄袭侵权/违法违规的内容， 请发送邮件至 chuangxiangniao@163.com 举报，一经查实，本站将立刻删除。
发布者：程序猿，转转请注明出处：https://www.chuangxiangniao.com/p/983216.html
微信扫一扫 
支付宝扫一扫 
