java - java惰性流构造中的问题

Question

我在看 this惰性流构建的链接，并尝试将其用于我的一个案例。

我的主流有一些需要在 Stream.onClose() 上完成的操作.

在我的自定义逻辑中，我使用来自 Stream.iterator() 的迭代器用于流处理。

这在消耗实际 Stream 时效果很好。但是，当我使用 Stream.flatMap()为了构建惰性流，onClose 函数在我开始迭代时被调用，这反过来又给我带来了问题。

我在 zulu-opendjk 1.8.0_222 和 13 中试过这个。我在这两个环境中都遇到了这个异常。

您可以使用以下代码重现该问题。

import java.util.*;
import java.util.stream.Stream;
import java.util.stream.StreamSupport;

public class TestStreamIterator
{
    public static void main(String args[])
    {
        Stream<String> stream1 = getStream();
        stream1.iterator().forEachRemaining(System.out::println);

        Stream<String> stream2 = Stream.of(1).flatMap(integer -> getStream());
        stream2.iterator().forEachRemaining(System.out::println);
    }

    private static Stream<String> getStream()
    {
        List<String> values = Arrays.asList("a", "b", "c");

        MyIterator iterator = new MyIterator(values);
        Stream<String> stream = StreamSupport.stream(Spliterators.spliteratorUnknownSize(iterator, Spliterator.NONNULL | Spliterator.IMMUTABLE), false).onClose(iterator::close);

        return stream;
    }

    private static class MyIterator implements Iterator<String>, AutoCloseable
    {
        private Iterator<String> iterator;

        public MyIterator(List<String> values)
        {
            iterator = values.iterator();
        }

        @Override
        public boolean hasNext()
        {
            return iterator.hasNext();
        }

        @Override
        public String next()
        {
            return iterator.next();
        }

        @Override
        public void close()
        {
            throw new IllegalStateException("Should not come here");
        }
    }
}

我的理解是，当使用 flatMap 时；只应调用 Stream.of(1) 的 close 方法。不是在 flatMap 函数中创建的流。

我原以为 onClose 函数仅在流关闭时被调用。但是，我不确定流在哪里关闭。

解决此案例的任何帮助也会有所帮助。

Answer 1

当您调用 flatMap(integer -> getStream()) 这里:

Stream<String> stream2 = Stream.of(1).flatMap(integer -> getStream());
stream2.iterator().forEachRemaining(System.out::println);

你正在调用这个方法:

Iterator

<R> Stream<R> flatMap(Function<? super T, ? extends Stream<? extends R>> mapper);

Returns a stream consisting of the results of replacing each element of this stream with the contents of a mapped stream produced by applying the provided mapping function to each element. Each mapped stream is closed after its contents have been placed into this stream. (If a mapped stream is null an empty stream is used, instead.)

因此，正如文档所述，您传递给此方法的映射流(来自 getStream()，它是 MyIterator 上的流)将被关闭，然后(如 onClose 中所定义)流)它调用MyIterator.close()抛出异常。

---

处理您的评论，因为您似乎没有关注:

Stream<String> stream2 = Stream.of(1).flatMap(integer -> getStream());

创建一个流，当您读取它时，它会延迟映射到子流的内容。当该子流被加载到主流时，子流将被关闭。

stream2.iterator().forEachRemaining(System.out::println);

你从主流读取，它映射到子流，子流读取所有子流然后关闭子流，子流然后调用 Stream.onClose()这叫MyIterator.close()