Java 8流中使用.peek()的方法

Question

我在使用流中的.peek()方法，但这种做法不被推荐，可惜我找不到其他解决方案。

static boolean fooAddTester(int size) {
    Foo foo = new foo(); //data structure

    return IntStream.range(0, size).
            .peek(i -> synchronized(r){foo.add(i)})
            .allMatch(e -> foo.isLegal());
}

我需要做的是遍历IntStream并在每次插入后检查foo数据结构是否合法。 这在逻辑上等同于：

static boolean fooAddTester(int size) {
    Foo foo = new foo(); //data structure

    for(int i=0; i<size; i++){
        foo.add(i);
        if(!foo.isLegal())
            return false;
    return true;
}

然而，这更加复杂，我正在尝试使用流来简化和学习。

不使用.peek()实现相同功能的方法是这样的：这确实有效 - 但我只是将问题移动到了.allMatch()中：

return IntStream.range(0, size).
            .allMatch(i -> {
                 synchronized(r){foo.add(i)};
                 foo.isLegal();
             )};

我的问题与这个问题非常相似，不同之处在于我每次都在检查，所以那些解决方案都不起作用。

所以我的问题是：

• .peek()只能用于调试吗？还是我可以用它来实现我的需求？
• 有更好的解决方案吗？
• 我应该使用第二种解决方案吗？

我正在寻找正确的解决方案，而不仅仅是一个可行的解决方案，所有那些代码都已经可行。

- Adam Shem-Ur

---

这可能会有所帮助：https://softwareengineering.stackexchange.com/questions/308977/is-it-an-antipattern-to-use-peek-to-modify-a-stream-element - Emil Hotkowski

synchronized 不会对你有太大的帮助 - 它只会保证一次只能将一个元素放入 foo; 但它不会保证 哪个 元素被放到 foo 中，因为在并行流情况下不存在处理顺序。 - Eugene

什么是 r？为什么您要在其上进行同步？您的流代码与循环不匹配。您正在对 int 值进行流处理，因此您需要一个静态方法 Foo.isLegal（int） 来使第一个片段被编译器接受。在最后一个片段中，您正在使用方法引用 Foo :: isLegal，而在此处不允许使用方法引用。显然，您从未尝试过这些替代方案。 - Holger

Foo::isLegal 的逻辑是什么？看着你的“逻辑等价”示例，可能有更好的逻辑解决方案来解决你的问题。 - SubOptimal

@Eugene，如果应该避免副作用，那我是不是就不应该使用流？ - Adam Shem-Ur

Answer 1

Stream#peek方法的文档如下，主要是为了支持调试，在流经管道中的某一点时想要查看元素，但这不是绝对的。

@Holger 给出了更确切的答案:

peek最有用的功能是查找处理过的流元素。

他在回答中还指出了一些副作用，peek操作取决于哪个终端操作被调用。因此，在内部使用peek时应当小心。

因此正确的方式只是使用for-each循环，因为Stream#collect不支持短路操作。

可选的方式是使用peek，因为你可以自己控制流。并且你需要移除synchronized块，在这里是不必要的。

return IntStream.range(0, size).peek(foo::add).allMatch(__ -> Foo.isLegal(foo));

- holi-java

---

使用peek最有用的功能是确定流元素是否已被处理，但要通过日志记录而不是依靠副作用。一旦添加了“parallel”，这种方法会以神秘的方式中断。同步将确保一次只处理一个元素，但即使如此，您仍然不知道处理的是哪个元素，缺乏处理顺序。 - Eugene

@Eugene，但我想说的是，原始问题的提出者可以控制如何创建、消费和操作流。所以这里没有问题。 - holi-java

我最初添加了synchronized以使用parallel()。非常感谢您的答复！ - Adam Shem-Ur

Answer 2

如果你真的想在逻辑之前使用其他流操作，我只能想到一种方法，但我不是很喜欢它...

 boolean result = true;
    try {
        IntStream.range(0, 10)
                .forEachOrdered(x -> {
                    foo.add(x);
                    if (!Foo.isLegal(foo)) {
                        throw new RuntimeException("just because");
                    }
                });
    } catch (RuntimeException re) {
        result = false;
    }

    System.out.println(result);

显然，您需要将 RuntimeException 替换为您自己的异常类型。

- Eugene