java - Java 8 Streams 中副作用的危险是什么？

Question

我试图理解我在 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 对有状态的 lamda 和副作用的警告。但我不明白为什么会出现此警告。

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

在某种程度上，我读到了这个:

If executed in parallel, the non-thread-safety of ArrayList would cause incorrect results, and adding needed synchronization would cause contention, undermining the benefit of parallelism.

但是谁能更清楚地帮助我理解警告？

Answer 1

来自 Javadoc:

Note also that attempting to access mutable state from behavioral parameters presents you with a bad choice with respect to safety and performance; if you do not synchronize access to that state, you have a data race and therefore your code is broken, but if you do synchronize access to that state, you risk having contention undermine the parallelism you are seeking to benefit from. The best approach is to avoid stateful behavioral parameters to stream operations entirely; there is usually a way to restructure the stream pipeline to avoid statefulness.

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

• 安全，因为您需要 Stream 试图最小化的同步
• 性能，因为所需的同步会花费您(在您的示例中，如果您使用 ConcurrentHashMap，这会产生费用)。

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

• 如果你想使用Stream和多线程流，你需要像myThings.parralelStream()那样使用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 不是线程安全的。

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

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

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

终结器(此处为 groupingBy)执行您正在执行的工作，并且可能会按顺序调用(我的意思是，Stream 可能会跨多个线程拆分，终结器可能会被调用多次(在不同的线程中)然后它可能需要合并。

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