Java lambdas,无状态lambdas和并行执行

Question

在尝试学习Java lambdas时,我遇到了一篇文章(下面列出),其中有关流API限制的部分,他指出:"有状态lambda在顺序执行时通常不是问题,但是在执行流时是并行的,它打破了".然后,他将此代码作为执行顺序导致的问题示例:

List<String> ss = ...;
List<String> result = ...;

Stream<String> stream = ss.stream();

stream.map(s -> {
    synchronized (result) {
      if (result.size() < 10) {
        result.add(s);
      }
    }
})
.forEach(e -> { });

我可以看到,如果它是并行化的,这将是非确定性的,但是我看不出你将如何使用无状态lambdas解决这个问题 - 是不是存在一些关于在列表中添加内容的固有的非确定性平行时尚.一个六岁的帽子可以理解的例子,也许是在C#中,将非常感激.

链接到原始文章http://blog.hartveld.com/2013/03/jdk-8-33-stream-api.html

Answer 1

我知道你在哪里暗示你的问题,我会尽力解释.

考虑一个由8个元素组成的输入列表:

[1, 2, 3, 4, 5, 6, 7, 8]

并且假设流将以下列方式并行化,实际上它们没有,并行的确切过程很难理解.
但是现在,假设他们将大小除以2,直到剩下两个元素.

分支部门看起来像这样:

1. 甲级:

   [1, 2, 3, 4]
[5, 6, 7, 8]

2. 第二师:

   [1, 2]
[3, 4]
[5, 6]
[7, 8]

现在我们有四个块(在我们的理论中)将由四个不同的线程处理,这些线程彼此不了解.
这确实可以通过同步一些外部资源来解决,但是你失去了并行化的好处,所以我们需要假设我们不同步,当我们不同步时,其他线程将看不到任何其他线程有什么完了,所以我们的结果将是垃圾.

现在关于无状态问题的部分问题,如何正确地并行处理？如何将以正确顺序并行处理的元素添加到列表中？

首先假设一个简单的映射函数,用lambda映射i -> i + 10,然后System.out::println在foreach中打印它.

现在在第二次分裂后,将发生以下情况:

[1, 2] -> [11, 12] -> { System.out.println(11); System.println(12); }
[3, 4] -> [13, 14] -> { System.out.println(13); System.println(14); }
[5, 6] -> [15, 16] -> { System.out.println(15); System.println(16); }
[7, 8] -> [17, 18] -> { System.out.println(17); System.println(18); }

除了由同一线程处理的所有元素(内部状态,不依赖)按顺序处理之外,不保证顺序.

如果你想按顺序处理它们,那么你需要使用forEachOrdered,这将确保所有线程以正确的顺序运行,并且你不会失去太多的并行化好处,因为它只适用于最终状态.

要了解如何将项目parelllized添加到列表中,请使用Collectors.toList()提供以下方法的方法来查看:

• 创建新列表.
• 将值添加到列表中.
• 合并两个列表.

现在,第二次分裂后将发生以下情况:

对于每四个线程,它将执行以下操作(此处仅显示一个线程):

1. 我们有[1, 2].
2. 我们将它映射到[11, 12].
3. 我们创建一个空的List<Integer>.
4. 我们添加11到列表中.
5. 我们添加12到列表中.

现在所有线程都已完成此操作,并且我们有四个包含两个元素的列表.

现在,以指定的顺序发生以下合并:

1. [11, 12] ++ [13, 14] = [11, 12, 13, 14]
2. [15, 16] ++ [17, 18] = [15, 16, 17, 18]
3. 最后 [11, 12, 13, 14] ++ [15, 16, 17, 18] = [11, 12, 13, 14, 15, 16, 17, 18]

因此,结果列表是有序的,并且映射已经并行完成.现在你也应该能够看到为什么parallallization需要更高的最小值只作为两个项目,否则新列表的创建和合并变得太昂贵.

我希望您现在明白为什么流操作应该是无状态的,以获得并行化的全部好处.