有状态和无状态 lambda 表达式有什么区别？

Question

根据 OCP 书，必须避免有状态操作，否则称为有状态 lambda 表达式。书中提供的定义是“有状态的 lambda 表达式是一个其结果取决于在管道执行期间可能发生变化的任何状态的表达式。”

他们提供了一个示例，其中并行流用于使用 .map() 函数将固定的数字集合添加到同步的 ArrayList 中。

arraylist 中的顺序是完全随机的，这应该让人看到有状态的 lambda 表达式在运行时会产生不可预知的结果。这就是为什么强烈建议在使用并行流时避免有状态操作，以消除任何潜在的数据副作用。

他们没有显示一个无状态的 lambda 表达式来解决相同的问题（将数字添加到同步的数组列表），我仍然不明白使用映射函数填充空的同步数组列表的问题是什么与数据......在管道执行过程中可能会改变的状态究竟是什么？他们指的是 Arraylist 本身吗？就像另一个线程决定将其他数据添加到 ArrayList 时，而 parallelstream 仍在添加数字并因此改变最终结果的过程中？

也许有人可以为我提供一个更好的例子来说明什么是有状态的 lambda 表达式以及为什么应该避免它。那将不胜感激。

谢谢

Answer 1

第一个问题是这样的：

 List<Integer> list = new ArrayList<>();

    List<Integer> result = Stream.of(1, 2, 3, 4, 5, 6)
            .parallel()
            .map(x -> {
                list.add(x);
                return x;
            })
            .collect(Collectors.toList());

System.out.println(list);

您不知道这里的结果是什么，因为您正在将元素添加到非线程安全集合ArrayList。

但即使你这样做：

  List<Integer> list = Collections.synchronizedList(new ArrayList<>());

并执行相同的操作，list 没有可预测的顺序。多个线程添加到此同步集合中。通过添加同步集合，您可以保证添加 所有元素（与普通的 ArrayList 不同），但它们将以 哪个 的顺序出现在未知中。

请注意，list 没有任何顺序可以保证一切，这称为处理顺序。而result 保证 是：[1, 2, 3, 4, 5, 6] 对于这个特定的例子。

根据问题，您通常可以摆脱stateful 操作；对于您的示例，返回 synchronized List 将是：

 Stream.of(1, 2, 3, 4, 5, 6)
            .filter(x -> x > 2) // for example a filter is present
            .collect(Collectors.collectingAndThen(Collectors.toList(), 
                          Collections::synchronizedList));

Answer 2

举个例子，让我们考虑下面的Consumer（注意：这样的功能的用处在这里无关紧要）：

public static class StatefulConsumer implements IntConsumer {

    private static final Integer ARBITRARY_THRESHOLD = 10;
    private boolean flag = false;
    private final List<Integer> list = new ArrayList<>();

    @Override
    public void accept(int value) {
        if(flag){   // exit condition
            return; 
        }
        if(value >= ARBITRARY_THRESHOLD){
            flag = true;
        }
        list.add(value); 
    }

}

它是一个消费者，它将向List 添加项目（我们不考虑如何取回列表或线程安全）并有一个标志（表示状态）。

这背后的逻辑是，一旦达到阈值，消费者就应该停止添加商品。

您的书试图说的是，因为没有保证函数必须使用Stream 的元素的顺序，所以输出是不确定的。

因此，他们建议您仅使用无状态函数，这意味着它们将始终使用相同的输入产生相同的结果。

Answer 3

这是一个有状态操作每次返回不同结果的示例：

public static void main(String[] args) {

Set<Integer> seen = new HashSet<>();

IntStream stream = IntStream.of(1, 2, 3, 1, 2, 3);

// Stateful lambda expression
IntUnaryOperator mapUniqueLambda = (int i) -> {
    if (!seen.contains(i)) {
        seen.add(i);
        return i;
    }
    else {
        return 0;
    }
};

int sum = stream.parallel().map(mapUniqueLambda).peek(i ->   System.out.println("Stream member: " + i)).sum();

System.out.println("Sum: " + sum);
}

就我而言，当我运行代码时，我得到了以下输出：

Stream member: 1
Stream member: 0
Stream member: 2
Stream member: 3
Stream member: 1
Stream member: 2
Sum: 9

如果我插入哈希集，为什么总和是 9？
答案：不同的线程占用了IntStream 的不同部分 例如，值 1 和 2 设法在不同的线程上结束。

Answer 4

有状态的 lambda 表达式 其结果取决于在管道执行期间可能发生变化的任何状态。在 另一方面，一个无状态的 lambda 表达式是一个结果 不依赖于在执行过程中可能改变的任何状态 管道。

来源：OCP: Oracle Certified Professional Java SE 8 Programmer II Study Guide: Exam 1Z0-809by Jeanne Boyarsky,‎ Scott Selikoff

    List < Integer > data = Collections.synchronizedList(new ArrayList < > ());

            Arrays.asList(1, 2, 3, 4, 5, 6, 7).parallelStream()


                   .map(i -> {
                    data.add(i);
                    return i;
                }) // AVOID STATEFUL LAMBDA EXPRESSIONS!
                .forEachOrdered(i -> System.out.print(i+" "));


            System.out.println();
            for (int e: data) {
                System.out.print(e + " ");

可能的输出：

1 2 3 4 5 6 7 
1 7 5 2 3 4 6 

强烈建议您在使用时避免有状态操作 并行流，以消除任何潜在的数据副作用。在 事实上，通常应该在串行流中避免使用它们 可能，因为它们会阻止您的流利用 并行化。

Answer 5

有状态 lambda 表达式的结果取决于在流管道执行期间可能发生变化的任何状态。

让我们在这里通过一个例子来理解这一点：

    List<Integer> list = Arrays.asList(1,2,3,4,5,6,7,8,9,10,11,12,13,14,15);
    List<Integer> result = new ArrayList<Integer>();

    list.parallelStream().map(s -> {
            synchronized (result) {
              if (result.size() < 10) {
                result.add(s);
              }
            }
            return s;
        }).forEach( e -> {});
     System.out.println(result);  

当您运行此代码 5 次时，输出将/可能一直不同。背后的原因是这里处理地图更新结果数组中的 Lambda 表达式。因为这里的结果数组取决于特定子流的数组大小，每次调用这个并行流时都会改变。

为了更好地理解并行流： 并行计算涉及将问题划分为子问题，同时解决这些问题（并行，每个子问题在单独的线程中运行），然后将解决方案的结果组合到子问题中。当流并行执行时，Java 运行时会将流划分为多个子流。聚合操作迭代并并行处理这些子流，然后组合结果。

希望对你有帮助！！！