Java 8 Streams 中副作用的危险是什么？答案

【问题标题】：What is the danger of side effects in Java 8 Streams?Java 8 Streams 中副作用的危险是什么？
【发布时间】：2017-10-31 17:22:50
【问题描述】：

我正在尝试理解我在 Streams 文档中发现的警告。我已经养成了使用 forEach() 作为通用迭代器的习惯。这导致我编写这种类型的代码：

public class FooCache {
    private static Map<Integer, Integer> sortOrderCache = new ConcurrentHashMap<>();
    private static Map<Integer, String> codeNameCache = new ConcurrentHashMap<>();

    public static void populateCache() {
        List<Foo> myThings = getThings();

        myThings.forEach(thing -> {
            sortOrderCache.put(thing.getId(), thing.getSortOrder());
            codeNameCache.put(thing.getId(), thing.getCodeName())
        });
    }
}

这是一个简单的例子。我了解此代码违反了 Oracle 对有状态 lamdas 和副作用的警告。但我不明白为什么会出现这个警告。

运行此代码时，它的行为似乎符合预期。那么我该如何打破它来证明为什么这是一个坏主意呢？

总的来说，我读到了这个：

如果并行执行，ArrayList 的非线程安全性将导致不正确的结果，并且添加所需的同步会导致争用，破坏了并行性的好处。

但是任何人都可以澄清一下以帮助我理解警告吗？

【问题讨论】：

你能提供警告吗？
您的示例中没有流。
有一个终端流操作叫forEach()，但是你连流都没有。你打电话给List.forEach()。
docs.oracle.com/javase/8/docs/api/java/lang/…
我不知道您所说的“别名”是什么意思，但不知道。这是一种无关的方法。事实上，它是由Iterable 接口声明的，它甚至没有流方法。

标签： java java-stream

【解决方案1】：

来自 Javadoc：

还要注意，尝试从行为访问可变状态参数为您提供了一个关于安全性和表现; 如果您不同步对该状态的访问，则您有数据竞争，因此您的代码已损坏，但如果您这样做同步对该状态的访问，您可能会遇到竞争破坏您正在寻求从中受益的并行性。最好的方法是避免有状态的行为参数完全流式操作；通常有一种方法可以重组流管道以避免状态。

这里的问题是，如果你访问一个可变状态，你会失去两个方面：

安全，因为您需要Stream 试图最小化的同步
性能，因为所需的同步会花费您（在您的示例中，如果您使用 ConcurrentHashMap，这是有成本的）。

现在，在您的示例中，这里有几点：

如果要使用Stream和多线程流，则需要使用parralelStream()，如myThings.parralelStream()；就目前而言，java.util.Collection 提供的forEach 方法很简单for each。
您将HashMap 用作static 成员并对其进行了变异。 HashMap 不是线程安全的；您需要使用ConcurrentHashMap。

在 lambda 和 Stream 的情况下，您不得更改流的源：

myThings.stream().forEach(thing -> myThings.remove(thing));

这可能会起作用（但我怀疑它会抛出 ConcurrentModificationException），但这可能不起作用：

myThings.parallelStream().forEach(thing -> myThings.remove(thing));

那是因为ArrayList 不是线程安全的。

如果您使用同步视图 (Collections.synchronizedList)，那么您将获得性能，因为您在每次访问时都进行了同步。

在您的示例中，您宁愿使用：

sortOrderCache = myThings.stream()
                         .collect(Collectors.groupingBy(
                           Thing::getId, Thing::getSortOrder);
codeNameCache= myThings.stream()
                       .collect(Collectors.groupingBy(
                         Thing::getId, Thing::getCodeName);

finisher（这里是groupingBy）完成你正在做的工作，可能会被顺序调用（我的意思是，Stream 可能会被拆分到多个线程中，finisher 可能会被调用多次（在不同的线程中）和那么它可能需要合并。

顺便说一句，您最终可能会删除 codeNameCache/sortOrderCache 并简单地存储 id->Thing 映射。

【讨论】：

【解决方案2】：

我相信文档中提到了以下代码所展示的副作用：

List<Integer> matched = new ArrayList<>();
List<Integer> elements = new ArrayList<>();

for(int i=0 ; i< 10000 ; i++) {
    elements.add(i);
}

elements.parallelStream()
    .forEach(e -> {
        if(e >= 100) {
            matched.add(e);
        }
    });
System.out.println(matched.size());

此代码并行流过列表，并尝试将符合特定条件的元素添加到其他列表中。由于结果列表未同步，执行上述代码时会得到java.lang.ArrayIndexOutOfBoundsException。

解决方法是创建一个新列表并返回，例如：

List<Integer> elements = new ArrayList<>();
for(int i=0 ; i< 10000 ; i++) {
    elements.add(i);
}   
List<Integer> matched = elements.parallelStream()
    .filter(e -> e >= 100)
    .collect(Collectors.toList());
System.out.println(matched.size());

【讨论】：

【解决方案3】：

副作用经常对状态和上下文做出假设。同时，您不能保证看到元素的特定顺序，并且多个线程可能同时运行。

除非您为此编写代码，否则这会产生非常微妙的错误，在尝试并行时很难跟踪和修复。

【讨论】：