1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53

    // 阻塞模式 查数据和业务逻辑交替执行
    private void fetchBatchWithDataResource(DataSource ds, String sql, String where, int fetchSize,
                                            Consumer<List<LinkedHashMap<String, Object>>> handle) {
        Connection connection = null;
        Statement stmt = null;
        ResultSet rs = null;
        try {
            connection = ds.getConnection();
            String query;
            if (StringUtils.isNotBlank(where)) {
                query = sql + " WHERE " + where;
            } else {
                query = sql;
            }

            log.info("stream export: query={}", query);

            stmt = connection.createStatement(ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY);
            stmt.setQueryTimeout(3600);
            stmt.setFetchSize(fetchSize);

            rs = stmt.executeQuery(query);
            int counter = 0;

            List<LinkedHashMap<String, Object>> data = new ArrayList<>();
            while (rs.next()) {
                ResultSetMetaData meta = rs.getMetaData();
                int columnCount = meta.getColumnCount();
                LinkedHashMap<String, Object> row = new LinkedHashMap<>();
                for (int i = 1; i <= columnCount; i++) {
                    row.put(meta.getColumnName(i), rs.getObject(i));
                }
                data.add(row);

                if (data.size() > fetchSize) {
                    handle.accept(data);
                    counter++;
                    data.clear();
                }
            }
            if (!data.isEmpty()) {
                handle.accept(data);
                counter++;
            }

            log.info("stream export: count={} dataSize={} ", counter, (counter - 1) * fetchSize + data.size());
        } catch (Exception e) {
            log.error("", e);
            throw new RuntimeException(e);
        } finally {
            close(connection, stmt, rs);
        }
    }