使用Java流使用数据库游标

Question

我想使用Java流使用数据库游标。我希望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;
...

它确实工作得很好，但是有点麻烦，我想利用流应用编程接口。

Answer 1

首先，我们必须讨论如何从数据库中获取数据。如果您的目的是浏览大量记录，而不希望在内存中同时加载所有记录，则有两种选择：

1. 把结果分页。
2. 让你的司机把结果分页。

如果您已经有了一个基于游标的迭代器，可以根据需要检索分页数据，那么您可以使用JDK API中的拆分器和流支持实用程序类将其转换为流。

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（））。

一旦有了迭代器，使用JDK API中的Spliterators和StreamSupport实用程序类从迭代器中构建流就非常简单了。

我的基本实现看起来有点像这样。这只是为了说明的目的。你可能想给你的特殊情况更多的爱。

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逻辑，因此当我们使用流时，必须确保执行流。close（）当我们完成它时，以确保我们关闭到目前为止保持活动状态的所有资源（即conn、stm和rs）。

最好的方法也许是使用资源的try，这样try将负责关闭流。

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

另一种方法是您自己对结果进行分页，这取决于数据库，但使用诸如limit和offset之类的select子句，您可以请求特定的记录页，处理它们，然后检索更多的记录。

select id from users LIMIT 1000 OFFSET 5

在这种情况下，迭代器将消耗所有页面，完成后，请求下一页，直到在最后一页中没有找到更多记录。

这种另一种方法的优点是资源可以在迭代器本身中立即控制。

我不会开发一个这样的例子，留给你去尝试。