java - 使用 Java Stream 使用数据库游标

Question

我想使用 Java Stream 使用数据库游标。我更喜欢 Java 流根据需要获取和处理行，并避免先在内存中加载所有 500 万行，然后再处理它们。
是否可以在不将整个表加载到 RAM 中的情况下使用它？
到目前为止，我的代码如下所示:

Cursor<Product> products = DAO.selectCursor(...);

// 1. Initialize variables
long count = 0;
...
for (Iterator<Product> it = products.iterator(); it.hasNext();) {
  Product p = it.next();
  // 2. Processing each row
  ...
}
// 3. Concluding (processing totals, stats, etc.)
double avg = total / count;
...

它运行良好，但有点麻烦，我想利用 Stream API。

Answer 1

首先，我们必须讨论如何从数据库中获取数据。如果您的目的是查看大量记录，并且不想一次将它们全部加载到内存中，则有两种选择:

对结果进行分页。

让您的驱动程序对结果进行分页。

如果您已经有一个基于 Cursor 的迭代器根据需要检索分页数据，然后您可以使用 Spliterators和 StreamSupport JDK API 中的实用程序类将其转换为 Stream .

Stream<Product> products = StreamSupport.stream(
                Spliterators.spliteratorUnknownSize(cursor.iterator(),
                        Spliterator.NONNULL |
                                Spliterator.ORDERED |
                                Spliterator.IMMUTABLE), false)

否则，您将不得不构建自己的东西。
驱动程序分页
如果您的 JDBC 驱动程序支持 fetch size 属性，您可以执行以下操作:

Connection con = ds.getConnection();
con.setAutoCommit(false);
PreparedStatement stm = con.prepareStatement("SELECT order_number FROM orders WHERE order_date >= '2018-08-12'", ResultSet.TYPE_FORWARD_ONLY);
stm.setFetchSize(1000);
ResultSet rs = stm.executeQuery();

此时， rs包含 1000 条记录的第一次提取，在您阅读上一页之前，它不会从数据库中检索更多记录。
所有这一切的棘手部分是，在读取完所有记录之前，您无法关闭任何资源(即连接、准备好的语句和结果集)，并且由于我们要构建的流默认情况下是惰性的，这意味着我们有保持所有这些资源打开，直到我们完成流。
也许最简单的方法是围绕这个逻辑构建一个迭代器，当迭代器真正到达所有数据的末尾时，你可以关闭所有资源(即 !rs.next() )或者另一种选择是在流已关闭 ( Stream.onClose() )。
一旦我们有了一个迭代器，使用 Spliterators 构建一个流是非常简单的。和 StreamSupport来自 JDK API 的实用程序类。
我的基本实现看起来有点像这样。这仅用于说明目的。您可能希望对您的特定案例给予更多的爱。

public Stream<String> getUsers() {
    DataSource ds = jdbcTemplate.getDataSource();
    try {
        Connection conn = ds.getConnection();
        conn.setAutoCommit(false);
        PreparedStatement stm = conn.prepareStatement("SELECT id FROM users", ResultSet.TYPE_FORWARD_ONLY);
        //fetch size is what guarantees only 1000 records at the time
        stm.setFetchSize(1000);
        ResultSet rs = stm.executeQuery();

        Iterator<String> sqlIter = new Iterator<>() {
            @Override
            public boolean hasNext() {
                try {
                    return rs.next();
                } catch (SQLException e) {
                    closeResources(conn, stm, rs);
                    throw new RuntimeException("Failed to read record from ResultSet", e);
                }
            }

            @Override
            public String next() {
                try {
                    return rs.getString("id");
                } catch (SQLException e) {
                    closeResources(conn, stm, rs);
                    throw new RuntimeException("Failed to read record from ResultSet", e);
                }
            }
        };

        //turn iterator into a stream
        return StreamSupport.stream(
                Spliterators.spliteratorUnknownSize(sqlIter,
                        Spliterator.NONNULL |
                                Spliterator.ORDERED |
                                Spliterator.IMMUTABLE), false
        ).onClose(() -> {
            //make sure to close resources when done with the stream
            closeResources(conn, stm, rs);
        });


    } catch (SQLException e) {
        logger.error("Failed to process data", e);
        throw new RuntimeException(e);
    }
}

private void closeResources(Connection conn, PreparedStatement ps, ResultSet rs) {
    try (conn; ps; rs) {
        logger.info("Resources successfully closed");
    } catch (SQLException e) {
        logger.warn("Failed to properly close database sources", e);
    }
}

这里的关键是要注意我们返回的流应该运行一些 onClose逻辑，所以当我们使用流时必须确保我们做一个 stream.close()当我们完成它以确保我们关闭到目前为止保持 Activity 的所有资源(即 conn 、 stm 和 rs )。
最好的方法也许是使用 try-with-resources，这样 try 将负责关闭流。

try(Stream<String> users = userRepo.getUsers()){
    //print users to the main output retrieving 1K at the time
    users.forEach(System.out::println);
}

手动分页
另一种方法是您自己对结果进行分页，这取决于数据库，但使用限制和偏移等选择子句，您可以请求特定的记录页面，处理它们，然后检索更多。

select id from users LIMIT 1000 OFFSET 5

在这种情况下，您的迭代器将消耗所有页面，并在完成后请求下一页，直到在最后一页中找不到更多记录。
这种另一种方法的优点是可以在迭代器本身中立即控制资源。
我不会开发一个这样的例子，让你去尝试。