java 8 parallelStream().forEach 结果数据丢失

Question

有两个测试用例使用parallelStream()：

List<Integer> src = new ArrayList<>();
for (int i = 0; i < 20000; i++) {
  src.add(i);
}
List<String> strings = new ArrayList<>();
       
src.parallelStream().filter(integer -> (integer % 2) == 0).forEach(integer -> strings.add(integer + ""));
    
System.out.println("=size=>" + strings.size());

=size=>9332

List<Integer> src = new ArrayList<>();
for (int i = 0; i < 20000; i++) {
  src.add(i);
}
List<String> strings = new ArrayList<>();

src.parallelStream().forEach(integer -> strings.add(integer + ""));

System.out.println("=size=>" + strings.size());

=size=>17908

为什么我在使用parallelStream时总是丢失数据？我做错了什么？

Answer 1

ArrayList不是线程安全的。你需要做

List<String> strings = Collections.synchronizedList(new ArrayList<>());

或者

List<String> strings = new Vector<>();

以确保所有更新同步，或切换到

List<String> strings = src.parallelStream()
    .filter(integer -> (integer % 2) == 0)
    .map(integer -> integer + "")
    .collect(Collectors.toList());

并将列表构建留给 Streams 框架。请注意，返回的列表是否可collect修改是未定义的，因此如果这是一个要求，您可能需要修改您的方法。

在性能方面，Stream.collect可能比使用Stream.forEach添加到同步集合要快得多，因为 Streams 框架可以在不同步的情况下单独处理每个线程中的值集合，并在最后以线程安全的方式组合结果。