java - 对于基于 I/O 的流，我应该在 flatMap 中使用 try-with-resource 吗？

Question

A Stream 是一个 AutoCloseable，如果基于 I/O，应该在 try-with-resource block 中使用。通过 flatMap() 插入的基于 I/O 的中间流怎么样？示例:

try (var foos = foos()) {
   return foos.flatMap(Foo::bars).toArray(Bar[]::new);
}

对比

try (var foos = foos()) {
  return foos.flatMap(foo -> {
    try (var bars = foo.bars()) {
      return bars;
    }
  }).toArray(Bar[]::new);
}

flatMap() 文档说:

Each mapped stream is closed after its contents have been placed into this stream.

嗯，这就是快乐的道路。如果中间发生异常怎么办？该流是否会保持未关闭状态并可能泄漏资源？那么我是否应该始终对中间流使用 try-with-resource？

Answer 1

像这样的构造是没有意义的

return foos.flatMap(foo -> {
    try (var bars = foo.bars()) {
        return bars;
    }
}).toArray(Bar[]::new);

因为这会在它返回给调用者之前关闭流，这使得子流完全无法使用。

事实上，函数的代码不可能确保关闭发生在函数之外的适当位置。这肯定是 API 设计者决定您不必这样做的原因，而 Stream 实现将负责。

这也适用于异常(exception)情况。 Stream 仍然确保在函数将其返回到 Stream 后关闭流:

try {
    IntStream.range(1, 3)
        .flatMap(i -> {
            System.out.println("creating "+i);
            return IntStream.range('a', 'a'+i)
                    .peek(j -> {
                        System.out.println("processing sub "+i+" - "+(char)j);
                        if(j=='b') throw new IllegalStateException();
                    })
                    .onClose(() -> System.out.println("closing "+i));
        })
        .forEach(i -> System.out.println("consuming "+(char)i));
} catch(IllegalStateException ex) {
    System.out.println("caught "+ex);
}

creating 1
processing sub 1 - a
consuming a
closing 1
creating 2
processing sub 2 - a
consuming a
processing sub 2 - b
closing 2
caught java.lang.IllegalStateException

您可以玩弄条件，看看构造的 Stream 总是关闭的。对于外层 Stream 中没有被处理的元素，根本就没有 Stream。

对于像 .flatMap(Foo::bars) 或 .flatMap(foo -> foo.bars()) 这样的 Stream 操作，你可以假设一旦 bars() 成功创建并返回一个 Stream，它会被确定地传递给调用者并正确关闭。

另一种情况是映射函数，它在 Stream 创建之后执行可能会失败的操作，例如

.flatMap(foo -> {
    Stream<Type> s = foo.bar();
    anotherOperation(); // Stream is not closed if this throws
    return s;
})

在这种情况下，有必要确保在异常(exception)情况下关闭，但仅在异常(exception)情况下:

.flatMap(foo -> {
    Stream<Type> s = foo.bar();
    try {
        anotherOperation();
    } catch(Throwable t) {
        try(s) { throw t; } // close and do addSuppressed if follow-up error
    }
    return s;
})

但显然，您应该遵循保持 lambda 简单的一般规则，在这种情况下您不需要这样的保护。