在管道中间关闭流

Question

当我执行此代码时,它会在流管道中打开大量文件:

public static void main(String[] args) throws IOException {
    Files.find(Paths.get("JAVA_DOCS_DIR/docs/api/"),
            100, (path, attr) -> path.toString().endsWith(".html"))
        .map(file -> runtimizeException(() -> Files.lines(file, StandardCharsets.ISO_8859_1)))
        .map(Stream::count)
        .forEachOrdered(System.out::println);
}

我得到一个例外:

java.nio.file.FileSystemException: /long/file/name: Too many open files

问题是Stream.count当流完成后不会关闭流.但鉴于它是终端操作,我不明白为什么不应该这样做.对于诸如reduce和之类的其他终端操作也是如此forEach.flatMap另一方面关闭它所包含的流.

该文档告诉我使用try-with-resouces语句在必要时关闭流.在我的情况下,我可以count用这样的东西替换线:

.map(s -> { long c = s.count(); s.close(); return c; } )

但这很嘈杂,并且在某些情况下可能会给大型复杂的管道带来真正的不便.

所以我的问题如下:

1. 为什么流设计不是为了让终端操作关闭他们正在处理的流？这将使它们在IO流中更好地工作.
2. 关闭流水线中IO流的最佳解决方案是什么？

---

runtimizeException是一种在RuntimeExceptions 中包装已检查异常的方法.

Answer 1

这里有两个问题:处理已检查的异常,例如IOException及时关闭资源.

预定义的功能接口都没有声明任何已检查的异常,这意味着它们必须在lambda中处理,或者包含在未经检查的异常中并重新抛出.看起来你的runtimizeException功能就是这样.您可能还必须为它声明自己的功能接口.你可能已经发现,这是一种痛苦.

在关闭像文件这样的资源时,有一些调查是在到达流末尾时自动关闭流.这很方便,但它不会在抛出异常时处理关闭.在流中没有神奇的做正确的机制.

我们留下了处理资源闭包的标准Java技术,即Java 7中引入的try-with-resources构造.TWR确实希望在打开调用堆栈时将资源关闭到同一级别."打开它的人必须关闭它"的原则适用.TWR还处理异常处理,这通常可以方便地在同一个地方处理异常处理和资源关闭.

在这个例子中,流有点不寻常,因为它将a映射Stream<Path>到a Stream<Stream<String>>.这些嵌套流是未关闭的流,当系统用完打开的文件描述符时,会导致最终异常.使这很困难的是文件通过一个流操作打开然后传递到下游; 这使得无法使用TWR.

构建此管道的另一种方法如下.

该Files.lines调用是打开文件的调用,因此必须是TWR语句中的资源.这个文件的处理是(某些)IOExceptions被抛出的地方,所以我们可以在同一个TWR语句中进行异常包装.这表明有一个简单的函数将路径映射到行计数,同时处理资源关闭和异常包装:

long lineCount(Path path) {
    try (Stream<String> s = Files.lines(path, StandardCharsets.ISO_8859_1)) {
        return s.count();
    } catch (IOException ioe) {
        throw new UncheckedIOException(ioe);
    }
}

一旦你有了这个辅助函数,主管道就像这样:

Files.find(Paths.get("JAVA_DOCS_DIR/docs/api/"),
           100, (path, attr) -> path.toString().endsWith(".html"))
     .mapToLong(this::lineCount)
     .forEachOrdered(System.out::println);

Answer 2

可以创建一个可靠地关闭管道中间的流的实用程序方法.

这确保了每个资源都使用try-with-resource-statement关闭,但是不需要自定义实用程序方法,并且比直接在lambda中编写try-statement要简单得多.

使用此方法,问题中的管道如下所示:

Files.find(Paths.get("Java_8_API_docs/docs/api"), 100,
        (path, attr) -> path.toString().endsWith(".html"))
    .map(file -> applyAndClose(
        () -> Files.lines(file, StandardCharsets.ISO_8859_1),
        Stream::count))
    .forEachOrdered(System.out::println);

实现如下:

/**
 * Applies a function to a resource and closes it afterwards.
 * @param sup Supplier of the resource that should be closed
 * @param op operation that should be performed on the resource before it is closed
 * @return The result of calling op.apply on the resource 
 */
private static <A extends AutoCloseable, B> B applyAndClose(Callable<A> sup, Function<A, B> op) {
    try (A res = sup.call()) {
        return op.apply(res);
    } catch (RuntimeException exc) {
        throw exc;
    } catch (Exception exc) {
        throw new RuntimeException("Wrapped in applyAndClose", exc);
    }
}

^{(由于需要关闭的资源在分配时经常会抛出异常,因此非运行时异常包含在运行时异常中,从而避免需要单独的方法来执行此操作.)}