Java Stream 生命周期回调答案

【问题标题】：Java Stream lifecycle callbacksJava Stream 生命周期回调
【发布时间】：2019-06-13 19:45:18
【问题描述】：

当流的最后一个元素已被处理并且流被完全“消耗”时，是否存在注册回调的优雅方法？

尤其是当流源（如 DB 游标、Iterator<T> 或自定义 Supplier<T>）是有限的并且可以明确知道何时没有更多数据时。

例子：

public Stream<Row> query(String sql){
   Connection c = dataSource.openConnection();
   Stream<Row> rows = MyDB.query(sql).stream();
   c.close();
   return rows;
}

现在，立即关闭连接是徒劳的，最好在流完全消耗时安排关闭连接。

我知道有 onClose() API，但这依赖于消费者在流上显式调用 close()。

【问题讨论】：

stackoverflow.com/a/34753512/2711488

标签： java java-stream

【解决方案1】：

调用onClose，并记录返回的流必须由调用者关闭。

/**
 * The returned stream must be closed.
 */
public Stream<Row> query(String sql){
    Connection c = dataSource.openConnection();
    return MyDB.query(sql).stream().onClose(() -> {
        try {
            c.close();
        } catch (SQLException e) {
            throw new RuntimeException(e);
        }
    });
}

【讨论】：

这是否意味着流的使用者必须在流上显式调用close()？例如，假设流是用Iterator 作为源构造的。那么，当源Iterator 的hasNext() 返回false 时会自动调用onClose 吗？
@S.D.是的，调用者必须调用 close 或使用 try-with-resources，就像调用者需要对任何其他资源绑定流（如 Files.lines() 返回的流）一样。没有自动“流用完”回调，因为流可能永远不会用完。
@ernest_k 因为终端操作可能不会完全消耗流。假设您有一个名为 s 的流和以下代码：s.limit(10).forEach(...)。这只消耗前 10 个值，因此您可以执行其他操作来处理其余的值。只有您知道您何时完成了流，因此关闭流是您的责任。
@Andreas 你知道任何允许“处理其余部分”的实际流源吗？
@Slaw 我想，语言太弱了，因为Stream 是一个接口。当您尝试多次使用它时，参考实现将引发异常。一个不同的东西可能是一个 source 可以返回一个新的流，但是，不能保证流只会从源中获取那些元素，它在概念上消耗，尤其是并行流进行预取。例如，BufferedReader 明确指出，一旦其lines() 流已被处理，它就处于未定义状态。所以这里没有“处理其余部分”。我不知道有什么反例。

【解决方案2】：

你想注册一个stream close handler：

public Stream<Row> query(String sql){
    Connection c = dataSource.openConnection();
    return MyDB.query(sql).stream().onClose(() -> c.close());
}

此方法的调用者必须明确关闭流！

（注意：我在这里省略了异常处理。）

【讨论】：

【解决方案3】：

一个主要工作的实用程序，除非返回的流没有迭代到最后：

public <T> Stream<T> wrap(Stream<T> stream, Runnable onEnd) {
    final Object endSignal = new Object();
    return Stream.concat(stream, Stream.of(endSignal))
            .peek(i -> {
                if(i == endSignal){onEnd.run();}
            })
            .filter(i -> i != endSignal)
            .map(i -> (T) i);
}

【讨论】：