Java非Spark应用中基于JDBC并行读取海量数据的标准方案问询

我完全懂你现在的困扰——单线程读取千万级MySQL记录导出CSV确实会慢到让人抓狂。你已经理清了并行处理的核心步骤，而且Spark的分区思路非常值得借鉴，其实我们可以把这个逻辑移植到普通Java应用里，核心就是把大查询拆成多个独立的小分片，用多线程并行处理，最后合并结果。下面我给你详细拆解可行的方案和具体代码实现：

一、核心思路：复刻Spark的分区策略 #

Spark通过partitionColumn、lowerBound、upperBound和numPartitions拆分查询的本质，是找一个分布均匀的数值型列（比如自增ID、时间戳），把数据范围切成N份，每个线程独立处理一份。在Java里我们可以手动实现这个逻辑，不需要依赖Spark框架。

二、具体实现步骤 #

1. 选好分区列是关键 #

优先选择数值型、分布均匀且带索引的列（比如自增主键id、转成时间戳的create_time），这样分片查询的效率最高，不会出现某片数据量远超其他分片的情况。如果没有合适的列，也可以用MOD(id, N)的方式分片，但性能会打折扣。

2. 计算分片范围 #

先执行一次轻量查询获取分区列的上下界：

// 获取id的最小/最大值，用于分片
String rangeQuery = "SELECT MIN(id), MAX(id) FROM users";
try (Connection conn = DriverManager.getConnection(DB_URL, DB_USER, DB_PASS);
     Statement stmt = conn.createStatement();
     ResultSet rs = stmt.executeQuery(rangeQuery)) {
    rs.next();
    long minId = rs.getLong(1);
    long maxId = rs.getLong(2);
}

再根据你想要的线程数（比如50个）计算每个分片的范围：

int numThreads = 50;
long totalRange = maxId - minId;
long chunkSize = totalRange / numThreads;
// 处理最后一个分片可能多出来的部分
if (totalRange % numThreads != 0) {
    chunkSize++;
}

3. 多线程并行读取与写入 #

这里要注意几个核心细节：

• 每个线程用独立的JDBC连接（JDBC Connection不是线程安全的，绝对不能共享）
• 避免多线程同时写入同一个文件导致冲突：可以让每个线程先写临时文件，最后再合并；或者用线程安全的写入组件
• 开启流式读取，避免把千万级数据一次性加载到内存导致OOM

下面是完整的并行实现示例：

import java.io.*;
import java.sql.*;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;

public class ParallelMySqlToCsv {
    // 配置信息
    private static final String DB_URL = "jdbc:mysql://localhost/test";
    private static final String DB_USER = "root";
    private static final String DB_PASS = "";
    private static final String TABLE = "users";
    private static final String PARTITION_COL = "id";
    private static final int NUM_THREADS = 50;
    private static final String OUTPUT_CSV = "users_export.csv";

    public static void main(String[] args) throws Exception {
        // 1. 获取分区列的上下界
        long minId, maxId;
        try (Connection conn = DriverManager.getConnection(DB_URL, DB_USER, DB_PASS);
             Statement stmt = conn.createStatement();
             ResultSet rs = stmt.executeQuery(String.format("SELECT MIN(%s), MAX(%s) FROM %s", PARTITION_COL, PARTITION_COL, TABLE))) {
            rs.next();
            minId = rs.getLong(1);
            maxId = rs.getLong(2);
        }

        // 2. 计算每个分片的范围
        long totalRange = maxId - minId;
        long chunkSize = totalRange / NUM_THREADS;
        if (totalRange % NUM_THREADS != 0) {
            chunkSize++;
        }

        // 3. 创建线程池，提交分片任务
        ExecutorService executor = Executors.newFixedThreadPool(NUM_THREADS);
        List<Future<File>> futures = new ArrayList<>();
        for (int i = 0; i < NUM_THREADS; i++) {
            long start = minId + i * chunkSize;
            long end = Math.min(start + chunkSize - 1, maxId);
            // 最后一个分片直接覆盖到maxId，避免遗漏
            if (i == NUM_THREADS - 1) {
                end = maxId;
            }
            futures.add(executor.submit(new DataExportTask(start, end)));
        }

        // 4. 合并所有临时文件到最终CSV
        try (BufferedWriter finalWriter = new BufferedWriter(new FileWriter(OUTPUT_CSV))) {
            // 先写入CSV表头
            finalWriter.write("id,first_name,last_name,date_created,is_admin,num_points\n");
            // 遍历每个线程的结果文件
            for (Future<File> future : futures) {
                File tempFile = future.get();
                try (BufferedReader tempReader = new BufferedReader(new FileReader(tempFile))) {
                    String line;
                    while ((line = tempReader.readLine()) != null) {
                        finalWriter.write(line);
                        finalWriter.newLine();
                    }
                }
                // 删除临时文件
                tempFile.delete();
            }
        }

        executor.shutdown();
        System.out.println("数据导出完成！");
    }

    // 每个线程执行的分片导出任务
    private static class DataExportTask implements Callable<File> {
        private final long startId;
        private final long endId;

        public DataExportTask(long startId, long endId) {
            this.startId = startId;
            this.endId = endId;
        }

        @Override
        public File call() throws Exception {
            // 创建临时文件存储当前分片的数据
            File tempFile = File.createTempFile("user_chunk_", ".csv");
            try (Connection conn = DriverManager.getConnection(DB_URL, DB_USER, DB_PASS);
                 Statement stmt = conn.createStatement(ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY);
                 // 开启流式读取，避免内存溢出（MySQL需要设置fetchSize为Integer.MIN_VALUE）
                 ResultSet rs = stmt.executeQuery(String.format("SELECT * FROM %s WHERE %s BETWEEN %d AND %d", TABLE, PARTITION_COL, startId, endId))) {
                stmt.setFetchSize(Integer.MIN_VALUE);

                try (BufferedWriter writer = new BufferedWriter(new FileWriter(tempFile))) {
                    while (rs.next()) {
                        int id = rs.getInt("id");
                        String firstName = rs.getString("first_name");
                        String lastName = rs.getString("last_name");
                        Date dateCreated = rs.getDate("date_created");
                        boolean isAdmin = rs.getBoolean("is_admin");
                        int numPoints = rs.getInt("num_points");
                        // 处理CSV转义（比如字段含逗号、引号的情况）
                        writer.write(String.format("%d,%s,%s,%s,%b,%d",
                                id, escapeCsv(firstName), escapeCsv(lastName),
                                dateCreated, isAdmin, numPoints));
                        writer.newLine();
                    }
                }
            }
            return tempFile;
        }

        // CSV字段转义工具方法
        private String escapeCsv(String value) {
            if (value == null) return "";
            if (value.contains(",") || value.contains("\"") || value.contains("\n")) {
                return "\"" + value.replace("\"", "\"\"") + "\"";
            }
            return value;
        }
    }
}

4. 关键优化点 #

• 流式读取：设置stmt.setFetchSize(Integer.MIN_VALUE)是MySQL JDBC驱动开启流式读取的关键，这样每次只会从数据库拉取少量数据，不会把千万级记录全加载到内存
• 独立连接：每个线程用自己的Connection，避免线程安全问题
• CSV转义：手动处理或使用OpenCSV等成熟库，避免生成的CSV格式错误
• 临时文件合并：比多线程直接写同一个文件更安全高效，表头只需要写入一次

三、其他可选方案 #

1. Spring环境下的简化实现：可以用JdbcTemplate配合TaskExecutor来简化多线程逻辑，核心分片思路不变
2. 分库分表场景：如果数据已经分库分表，可以直接让每个线程处理一个分表，无需计算范围
3. Apache Commons DBUtils：用AsyncQueryRunner简化异步JDBC查询，但仍需自行拆分查询范围

对比你原来的单线程代码，并行方案的优势很明显：充分利用多核CPU的能力，每个分片的查询压力更小，同时流式读取避免了内存溢出的风险，导出速度会有量级的提升。

内容的提问来源于stack exchange，提问作者M80