在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()方法时，可能会发生以下问题：
立即学习“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. 
