java - 并行流将项目添加到同一数组

Question

我有一个对象数组，我流式传输以在每个对象上使用一个函数来修改它并将其添加到一个全新的对象数组。

通过使用 .parallel()，执行时间加快了 2 倍，但在遍历新数组时我得到了一些 NullPointerException。

调试了几次都没有成功。似乎这个问题只发生在运行时。我尝试实现一个用于将新对象添加到列表中的同步函数，但遗憾的是它并没有正常工作。

任何人都可以给我建议如何让它工作吗？提前致谢!

这是代码片段:

private static final Object sync = new Object();
private ArrayList<Object> newList = new ArrayList<Object>();      

private void addNewObject(Object newObject) {
        synchronized (sync) {
            newList.add(newObject);
        }
    }

private Object mutateObject(Object oldObject) {
    // Do something with the object here
    return mutatedObject;
}

public ArrayList<Object> createNewList(ArrayList<Object> oldList) {
    oldList.stream().parallel().forEach(object -> addNewObject(mutateObject(object)));
    return newList;
}

Answer 1

考虑将 createNewList 方法重构为如下内容:

public List<Object> createNewList(List<Object> oldList) {
    return oldList.parallelStream()
            .map(this::mutateObject)
            .collect(Collectors.toList());
}

如果您查看 java.util.stream.Stream JavaDoc你可以看到有一些终端操作，比如forEach、collect、toArray等。在你的问题中最好使用collect 终端操作，因为它在内部处理同步，避免了竞争条件情况，而无需进一步实现同步。

使用 .map() 映射对象(Stream 的中间操作)并将结果并行收集到最终列表解决了这个问题。

应该删除 addNewObject 函数和 sync 对象，因为它不再被使用。