注册流"完成"钩子

Question

使用Java 8 StreamAPI,我想注册一个"完成钩子",类似于:

Stream<String> stream = Stream.of("a", "b", "c");

// additional filters / mappings that I don't control
stream.onComplete((Completion c) -> {
    // This is what I'd like to do:
    closeResources();

    // This might also be useful:
    Optional<Throwable> exception = c.exception();
    exception.ifPresent(e -> throw new ExceptionWrapper(e));
});

我为什么要那么做的原因是因为我想包装在一个资源Stream的API客户端,消费,我想的是Stream,一旦它被消耗自动清理资源.如果可能,那么客户可以致电:

Collected collectedInOneGo =
Utility.something()
       .niceLookingSQLDSL()
       .moreDSLFeatures()
       .stream()
       .filter(a -> true)
       .map(c -> c)
       .collect(collector);

而不是目前所需要的:

try (Stream<X> meh = Utility.something()
                            .niceLookingSQLDSL()
                            .moreDSLFeatures()
                            .stream()) {

    Collected collectedWithUglySyntacticDissonance =
    meh.filter(a -> true)
       .map(c -> c)
       .collect(collector);
}

理想情况下,我想进入java.util.stream.ReferencePipeline各种方法,例如:

@Override
final void forEachWithCancel(Spliterator<P_OUT> spliterator, Sink<P_OUT> sink) {
    try {

        // Existing loop
        do { } while (!sink.cancellationRequested() && spliterator.tryAdvance(sink));
    }

    // These would be nice:
    catch (Throwable t) {
        completion.onFailure(t);
    }
    finally {
        completion.onSuccess();
    }
}

使用现有的JDK 8 API有一种简单的方法吗？

Answer 1

最简单的解决方案是将流包装在另一个流中并将其平面映射到自身:

// example stream
Stream<String> original=Stream.of("bla").onClose(()->System.out.println("close action"));

// this is the trick
Stream<String> autoClosed=Stream.of(original).flatMap(Function.identity());

//example op
int sum=autoClosed.mapToInt(String::length).sum();
System.out.println(sum);

它起作用的原因在于flatMap操作:

每个映射的流在其内容放入此流后关闭.

但是当前的实现并不像使用时那样懒惰flatMap.这已在Java 10中修复.

我的建议是在try(…)需要关闭返回的流时继续使用标准解决方案和文档.毕竟,在终端操作之后关闭资源的流是不安全的,因为没有保证客户端实际上将调用终端操作.改变它的想法并放弃流即时是一种有效的用途,而close()当文档指定它是必需的时,不调用该方法则不是.

Answer 2

拦截终端操作除了flatMap基于解决方案(由@Holger提出)的任何解决方案都会对以下代码脆弱:

Stream<String> stream = getAutoCloseableStream();
if(stream.iterator().hasNext()) {
    // do something if stream is non-empty
}

这种用法在规范中绝对合法.不要忘记它iterator()并且spliterator()是终端流操作,但在执行后您仍然需要访问流源.放弃Iterator或Spliterator处于任何状态都是完全有效的,所以你无法知道它是否会被进一步使用.

您可以考虑advicing用户不要使用iterator()和spliterator(),但对于这个代码？

Stream<String> stream = getAutoCloseableStream();
Stream.concat(stream, Stream.of("xyz")).findFirst();

这内部spliterator().tryAdvance()用于第一个流,然后放弃它(尽管在close()显式调用结果流时关闭).您需要让用户不要使用Stream.concat.而据我了解内部库中的您正在使用iterator()/ spliterator()漂亮的时候,所以你需要重新审视这些地方可能出现的问题.而且,当然还有很多其他库也使用iterator()/ spliterator()并且可能在此之后短路:所有这些库都会与您的功能不兼容.

为什么flatMap基于解决方案在这里工作？因为在的第一个呼叫hasNext()或者tryAdvance()它转储整个流内容到中间缓冲器和关闭所述原始流源.因此,根据流的大小,您可能会浪费很多中间内存,甚至可能OutOfMemoryError.

您也可以考虑将PhantomReferences 保留在Stream对象上并监视对象ReferenceQueue.在这种情况下,完成将由垃圾收集器触发(这也有一些缺点).

总之,我的建议是继续尝试资源.

Answer 3

Java 8已经开创了需要关闭的流如何运行的先例.在他们的Javadoc中,它提到:

Streams有一个BaseStream.close()方法并实现AutoCloseable,但几乎所有的流实例实际上都不需要在使用后关闭.通常,只有源为IO通道的流(例如Files.lines(Path,Charset)返回的流)才需要关闭.大多数流都由集合,数组或生成函数支持,不需要特殊的资源管理.(如果流确实需要关闭,则可以在try-with-resources语句中将其声明为资源.)

所以Java 8的建议是在try-with-resources中打开这些流.一旦你这样做,Stream 还提供了一种方法来添加一个关闭钩子,几乎完全如你所描述的那样:onClose(Runnable)它接受一个lambda告诉它该做什么并返回一个Stream也会在关闭时执行该操作.

这就是API设计和文档建议你做的事情的方式.