检查java流是否已被消费

Question

如何检查流实例是否已被使用（意味着调用了终端操作，这样对终端操作的任何进一步调用都可能因IllegalStateException: stream has already been operated upon or closed. 而失败？

理想情况下，我想要一个在流尚未被消费时不消费流的方法，如果流已被消费而没有从流方法中捕获IllegalStateException，则返回布尔值 false（因为使用异常进行控制flow 成本高且容易出错，尤其是在使用标准异常时）。

在异常抛出和布尔返回行为中类似于迭代器中的hasNext() 的方法（尽管没有与next() 的约定）。

例子：

public void consume(java.util.function.Consumer<Stream<?>> consumer, Stream<?> stream) {
   consumer.accept(stream);
   // defensive programming, check state
   if (...) {
       throw new IllegalStateException("consumer must call terminal operation on stream");
   }
}

如果客户端代码调用此方法而不使用流，目标是尽早失败。

似乎没有办法做到这一点，我必须添加一个 try-catch 块来调用任何终端操作，如 iterator()，捕获一个异常并抛出一个新异常。

可接受的答案也可以是“不存在解决方案”，并有充分的理由说明为什么规范不能添加这种方法（如果存在充分的理由）。似乎 JDK 流通常在其终端方法的开头有这个 sn-ps：

// in AbstractPipeline.java
if (linkedOrConsumed)
    throw new IllegalStateException(MSG_STREAM_LINKED);

所以对于那些流来说，实现这样的方法似乎并不难。

Answer 1

考虑到spliterator（例如）是一个终端操作，你可以简单地创建一个类似的方法：

private static <T> Optional<Stream<T>> isConsumed(Stream<T> stream) {

    Spliterator<T> spliterator;
    try {
        spliterator = stream.spliterator();
    } catch (IllegalStateException ise) {
        return Optional.empty();
    }

    return Optional.of(StreamSupport.stream(
        () -> spliterator,
        spliterator.characteristics(),
        stream.isParallel()));
}

我不知道有什么更好的方法……而且用法是：

Stream<Integer> ints = Stream.of(1, 2, 3, 4)
                                 .filter(x -> x < 3);

YourClass.isConsumed(ints)
         .ifPresent(x -> x.forEachOrdered(System.out::println));

由于我认为没有实际理由返回已使用的 Stream，因此我将返回 Optional.empty()。

Answer 2

一种解决方案是在将中间操作（例如filter()）传递给stream 之前，将其传递给consumer。在该操作中，除了保存调用该操作的状态（例如，使用AtomicBoolean）之外，您什么都不做：

public <T> void consume(Consumer<Stream<T>> consumer, Stream<T> stream) {
    AtomicBoolean consumed = new AtomicBoolean(false);
    consumer.accept(stream.filter(i -> {
        consumed.set(true);
        return true;
    }));
    if (!consumed.get()) {
        throw new IllegalStateException("consumer must call terminal operation on stream");
    }
}

旁注：不要为此使用peek()，因为它不会被短路终端操作调用（如findAny()）。

Answer 3

这是一个独立的可编译解决方案，它使用委托自定义 Spliterator<T> 实现 + AtomicBoolean 来完成您所寻求的，而不会丢失线程安全或影响 Stream<T> 的并行性。

主要入口是Stream<T> track(Stream<T> input, Consumer<Stream<T>> callback) 函数——你可以在回调函数中做任何你想做的事情。我首先修改了一个委托Stream<T> 实现，但它的接口太大而无法委托而没有任何问题（请参阅我的代码注释，即使Spliterator<T> 在委托时也有其警告）：

import java.util.Spliterator;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Consumer;
import java.util.stream.IntStream;
import java.util.stream.Stream;
import java.util.stream.StreamSupport;

class StackOverflowQuestion56927548Scratch {

    private static class TrackingSpliterator<T> implements Spliterator<T> {
        private final AtomicBoolean tracker;
        private final Spliterator<T> delegate;
        private final Runnable callback;

        public TrackingSpliterator(Stream<T> forStream, Runnable callback) {
            this(new AtomicBoolean(true), forStream.spliterator(), callback);
        }

        private TrackingSpliterator(
                AtomicBoolean tracker,
                Spliterator<T> delegate,
                Runnable callback
        ) {
            this.tracker = tracker;
            this.delegate = delegate;
            this.callback = callback;
        }

        @Override
        public boolean tryAdvance(Consumer<? super T> action) {
            boolean advanced = delegate.tryAdvance(action);
            if(tracker.compareAndSet(true, false)) {
                callback.run();
            }
            return advanced;
        }

        @Override
        public Spliterator<T> trySplit() {
            Spliterator<T> split = this.delegate.trySplit();
            //may return null according to JavaDoc
            if(split == null) {
                return null;
            }
            return new TrackingSpliterator<>(tracker, split, callback);
        }

        @Override
        public long estimateSize() {
            return delegate.estimateSize();
        }

        @Override
        public int characteristics() {
            return delegate.characteristics();
        }
    }

    public static <T> Stream<T> track(Stream<T> input, Consumer<Stream<T>> callback) {
        return StreamSupport.stream(
                new TrackingSpliterator<>(input, () -> callback.accept(input)),
                input.isParallel()
        );
    }

    public static void main(String[] args) {
        //some big stream to show it works correctly when parallelized
        Stream<Integer> stream = IntStream.range(0, 100000000)
                .mapToObj(Integer::valueOf)
                .parallel();
        Stream<Integer> trackedStream = track(stream, s -> System.out.println("consume"));

        //dummy consume
        System.out.println(trackedStream.anyMatch(i -> i.equals(-1)));
    }
}

只需返回track 函数的流，也许调整callback 参数类型（您可能不需要传递流），您就可以开始了。

请注意，此实现仅跟踪实际使用流的时间，在由例如生成的 Stream 上调用 .count()。 IntStream.range(0,1000)（没有任何过滤步骤等）不会消耗流，而是通过Spliterator<T>.estimateSize()返回流的底层已知长度！