java.util.stream 与 ResultSet

Question

我很少有包含大量数据的表(大约 1 亿条记录)。所以我无法将这些数据存储在内存中，但我想使用 java.util.stream 类流式传输这个 result set 并将这个流传递给另一个类。我阅读了 Stream.of 和 Stream.Builder 运算符，但它们是内存中的缓冲流。那么有什么办法可以解决这个问题吗？提前致谢。

更新 #1

好的，我用谷歌搜索并找到了 jooq 库。我不确定，但看起来它可能适用于我的测试用例。总而言之，我很少有包含大量数据的表。我想流式传输我的结果集并将此流传输到另一个方法。像这样的:

// why return Stream<String>? Because my result set has String type
private Stream<Record> writeTableToStream(DataSource dataSource, String table) {

    Stream<Record> record = null;
    try (Connection connection = dataSource.getConnection()) {
        String sql = "select * from " + table;

        try (PreparedStatement pSt = connection.prepareStatement(sql)) {
            connection.setAutoCommit(false);
            pSt.setFetchSize(5000);
            ResultSet resultSet = pSt.executeQuery();
            //
            record = DSL.using(connection)
                    .fetch(resultSet).stream();
        }
    } catch (SQLException sqlEx) {
        logger.error(sqlEx);
    }

    return record;
}

请有人建议，我的方法是否正确？谢谢。

更新 #2

我在 jooq 上做了一些实验，现在可以说上面的决定不适合我。这段代码 record = DSL.using(connection).fetch(resultSet).stream(); 耗时太多

Answer 1

你首先要明白的就是这样的代码

try (Connection connection = dataSource.getConnection()) {
    …
    try (PreparedStatement pSt = connection.prepareStatement(sql)) {
        …
        return stream;
    }
}

在您离开 try 时不起作用 block ，资源在处理 Stream 时关闭还没开始呢。

资源管理构造“try with resources”适用于方法内的 block 范围内使用的资源，但您正在创建返回资源的工厂方法。因此，您必须确保关闭返回的流将关闭资源，并且调用者负责关闭 Stream .

---

此外，您需要一个从 ResultSet 中的单行生成项目的函数。 .假设，你有一个类似的方法

Record createRecord(ResultSet rs) {
    …
}

您可以创建一个 Stream<Record>基本喜欢

Stream<Record> stream = StreamSupport.stream(new Spliterators.AbstractSpliterator<Record>(
    Long.MAX_VALUE,Spliterator.ORDERED) {
        @Override
        public boolean tryAdvance(Consumer<? super Record> action) {
            if(!resultSet.next()) return false;
            action.accept(createRecord(resultSet));
            return true;
        }
    }, false);

但是要正确地做到这一点，您必须结合异常处理和资源关闭。您可以使用Stream.onClose注册将在 Stream 时执行的操作关闭，但必须是 Runnable不能抛出已检查的异常。同样的 tryAdvance方法不允许抛出已检查的异常。因为我们不能简单地嵌套 try(…)此处阻塞，close中抛出的抑制异常的程序逻辑，当已经有一个挂起的异常时，不是免费的。

为了帮助我们，我们引入了一种新类型，它可以包装关闭操作，这些操作可能会抛出已检查的异常并将它们包装在未检查的异常中。通过实现 AutoCloseable本身，它可以利用 try(…)构造以安全地链接关闭操作:

interface UncheckedCloseable extends Runnable, AutoCloseable {
    default void run() {
        try { close(); } catch(Exception ex) { throw new RuntimeException(ex); }
    }
    static UncheckedCloseable wrap(AutoCloseable c) {
        return c::close;
    }
    default UncheckedCloseable nest(AutoCloseable c) {
        return ()->{ try(UncheckedCloseable c1=this) { c.close(); } };
    }
}

这样，整个操作就变成了:

private Stream<Record> tableAsStream(DataSource dataSource, String table)
    throws SQLException {

    UncheckedCloseable close=null;
    try {
        Connection connection = dataSource.getConnection();
        close=UncheckedCloseable.wrap(connection);
        String sql = "select * from " + table;
        PreparedStatement pSt = connection.prepareStatement(sql);
        close=close.nest(pSt);
        connection.setAutoCommit(false);
        pSt.setFetchSize(5000);
        ResultSet resultSet = pSt.executeQuery();
        close=close.nest(resultSet);
        return StreamSupport.stream(new Spliterators.AbstractSpliterator<Record>(
            Long.MAX_VALUE,Spliterator.ORDERED) {
            @Override
            public boolean tryAdvance(Consumer<? super Record> action) {
                try {
                    if(!resultSet.next()) return false;
                    action.accept(createRecord(resultSet));
                    return true;
                } catch(SQLException ex) {
                    throw new RuntimeException(ex);
                }
            }
        }, false).onClose(close);
    } catch(SQLException sqlEx) {
        if(close!=null)
            try { close.close(); } catch(Exception ex) { sqlEx.addSuppressed(ex); }
        throw sqlEx;
    }
}

此方法包装了所有资源的必要关闭操作，Connection , Statement和 ResultSet在上述实用程序类的一个实例中。如果在初始化过程中发生异常，则立即执行关闭操作并将异常传递给调用者。如果流构造成功，关闭操作通过onClose注册。 .

因此调用者必须确保正确关闭

try(Stream<Record> s=tableAsStream(dataSource, table)) {
    // stream operation
}

请注意，SQLException通过RuntimeException已添加到 tryAdvance方法。因此，您现在可以添加 throws SQLException到createRecord方法没有问题。