Java多线程并行读取数据库百万级记录的优化方案咨询

一、多线程实现方案（避免重复读取+保障一致性） #

1. 避免重复读取的核心思路 #

• 主键范围拆分：根据源数据的主键（或唯一有序字段）将百万条记录划分为N个连续区间，每个线程负责一个区间的读取和写入。比如将ID从1到100万分成10个区间，每个线程处理10万条，直接通过WHERE id BETWEEN start AND end查询，从根源避免重复。
• 生产者-消费者模式：单线程作为生产者，负责分页读取源数据并放入阻塞队列；多个消费者线程从队列中取数据写入目标系统。由生产者统一控制读取进度，确保数据不重复、不遗漏。

2. 数据一致性保障措施 #

• 分块事务提交：每个线程处理完一个数据块后再提交事务，若某块写入失败，仅需重试当前块，无需全量回滚。
• 幂等性设计：目标系统的插入操作需支持幂等，比如以业务唯一键或主键作为判断依据，重复插入时自动忽略或报错，避免生成重复数据。
• 断点续传：用日志表或本地文件记录已处理完成的数据区间/页码，任务重启时可从断点继续执行，无需从头开始。

3. Java代码示例 #

主键范围拆分实现

// 初始化固定大小线程池，根据CPU核心数或IO能力调整
ExecutorService executor = Executors.newFixedThreadPool(8);

// 获取源数据主键的最小/最大值
long minId = getMinPrimaryKey();
long maxId = getMaxPrimaryKey();
// 计算每个线程处理的数据量
long batchSize = (maxId - minId) / 8 + 1;

// 提交分区间任务
for (long start = minId; start <= maxId; start += batchSize) {
    long end = Math.min(start + batchSize - 1, maxId);
    executor.submit(() -> {
        // 读取当前区间的数据
        List<BusinessData> dataList = queryDataByRange(start, end);
        // 批量写入目标系统
        batchInsertToTargetSystem(dataList);
    });
}

// 等待所有任务完成
executor.shutdown();
try {
    executor.awaitTermination(1, TimeUnit.HOURS);
} catch (InterruptedException e) {
    Thread.currentThread().interrupt();
}

生产者-消费者模式实现

// 阻塞队列，缓冲读取的数据块，容量根据内存调整
BlockingQueue<List<BusinessData>> dataQueue = new ArrayBlockingQueue<>(10);

// 生产者线程：分页读取源数据
Thread producerThread = new Thread(() -> {
    int pageNum = 1;
    final int PAGE_SIZE = 1000;
    while (true) {
        List<BusinessData> dataList = queryDataByPage(pageNum++, PAGE_SIZE);
        if (dataList.isEmpty()) break;
        try {
            dataQueue.put(dataList);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            break;
        }
    }
    // 放入空列表作为结束标记
    try {
        dataQueue.put(Collections.emptyList());
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
    }
});

// 消费者线程池
ExecutorService consumerExecutor = Executors.newFixedThreadPool(8);
for (int i = 0; i < 8; i++) {
    consumerExecutor.submit(() -> {
        while (true) {
            try {
                List<BusinessData> dataList = dataQueue.take();
                if (dataList.isEmpty()) {
                    // 放回结束标记，通知其他消费者
                    dataQueue.put(dataList);
                    break;
                }
                // 批量写入目标系统
                batchInsertToTargetSystem(dataList);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                break;
            }
        }
    });
}

// 启动并等待任务完成
producerThread.start();
try {
    producerThread.join();
} catch (InterruptedException e) {
    Thread.currentThread().interrupt();
}
consumerExecutor.shutdown();
try {
    consumerExecutor.awaitTermination(1, TimeUnit.HOURS);
} catch (InterruptedException e) {
    Thread.currentThread().interrupt();
}

二、非多线程优化方案 #

• 批量读写优化：放弃逐条操作，改为批量读取（设置JDBC的fetchSize为合适值，如1000）和批量写入（使用JDBC的addBatch()/executeBatch()，或调用目标系统的批量接口），这是最基础且见效明显的优化。
• 数据库原生迁移工具：如果源库和目标库类型匹配，直接使用数据库自带的迁移工具，比如MySQL的mysqldump、Oracle的expdp/impdp，效率远高于Java代码读写。
• CDC实时同步：采用变更数据捕获工具（如Debezium），实时同步源库数据到目标系统，无需每日日终批量跑任务，彻底解决日终压力。
• 消息队列解耦：将源数据写入Kafka等消息队列，目标系统异步消费队列数据并插入，源系统无需等待写入完成，提升整体流程效率。
• 数据压缩传输：若跨网络传输数据，将读取的数据序列化后压缩（如Gzip），减少网络IO耗时。

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