基于生产-消费模型的批量导出

设计思想

将“查询数据”和“写入 Excel”解耦，查询端可并发获取数据，写入端保持单线程顺序写入，避免写并发造成的文件损坏风险。
使用阻塞队列在生产者与消费者之间传递批次数据，平衡数据库读取吞吐与写入吞吐。
生产者数量依据数据量动态调整，避免小数据集过度并发，避免大数据集单线程瓶颈。
允许导出不保证顺序，以更高吞吐为目标。

优点

速度快：多生产者并行批量查询，充分利用数据库和 CPU。
写入安全：消费者单线程写 Excel，避免线程间写冲突。
吞吐稳定：队列缓冲削峰填谷，生产/消费速度不一致时也能稳定运行。
可扩展：生产者策略、页大小、队列大小可配置调整。

与传统批量导出的对比

维度	传统单线程分页导出	生产-消费并发导出
查询并发	单线程顺序查询	多线程并行查询
写入方式	单线程顺序写	单线程顺序写
顺序保证	有	无（不要求顺序）
吞吐上限	受单线程限制	受数据库与 IO 共同限制
稳定性	简单但慢	稍复杂但更快
适用场景	数据量较小	数据量较大、追求速度

代码实现细节

1) 接口层：无参数的快速导出入口

/trade/export/fast 不接收参数，直接触发快速导出。

@GetMapping(value = "/export/fast")
public ResponseEntity<StreamingResponseBody> exportFast() {
    ExportConfig config = new ExportConfig();
    String filename = "trade_fast_" + LocalDateTime.now().format(FILE_TS) + ".xlsx";

    StreamingResponseBody body = outputStream -> tradeExportService.exportFast(
            outputStream,
            config
    );

    return ResponseEntity.ok()
            .contentType(MediaType.APPLICATION_OCTET_STREAM)
            .header("Content-Disposition", "attachment; filename=\"" + filename + "\"")
            .body(body);
}

2) 服务层：多生产者 + 单消费者

先统计总量，计算总页数。
动态计算生产者数量。
生产者并发分页查询（基于 offset），把批次数据放入队列。
消费者单线程从队列中取数据写入 Excel。
通过“毒丸”标识生产者结束，消费者收齐后退出。

public void exportFast(OutputStream out, ExportConfig config) {
    long total = tradeMapper.countForExport();
    int pageSize = config.getPageSize();
    int totalPages = (int) ((total + pageSize - 1) / pageSize);
    int resolvedProducers = resolveProducerCount(totalPages);

    BlockingQueue<ExportBatch> queue =
            new ArrayBlockingQueue<>(resolvedProducers * 2);
    ExecutorService executor = Executors.newFixedThreadPool(resolvedProducers);
    AtomicInteger pageIndex = new AtomicInteger(0);

    // 生产者
    for (int i = 0; i < resolvedProducers; i++) {
        executor.submit(() -> {
            while (true) {
                int page = pageIndex.getAndIncrement();
                if (page >= totalPages) {
                    break;
                }
                long offset = (long) page * pageSize;
                List<TradeExportVO> rows = tradeMapper.selectByOffset(offset, pageSize);
                if (rows != null && !rows.isEmpty()) {
                    queue.put(new ExportBatch(rows));
                }
            }
            queue.put(ExportBatch.poison());
        });
    }

    // 消费者
    ExcelWriter writer = EasyExcelFactory.write(out, TradeExportVO.class).build();
    int poisonReceived = 0;
    while (poisonReceived < resolvedProducers) {
        ExportBatch batch = queue.take();
        if (batch.poison) {
            poisonReceived++;
            continue;
        }
        writer.write(batch.rows, sheet);
    }
    writer.finish();
}

3) 动态生产者数量策略

以页数为基础，设置阶梯式并发上限，同时受 CPU 核心数限制。

private static int resolveProducerCount(int totalPages) {
    int cpu = Runtime.getRuntime().availableProcessors();
    if (totalPages <= 1) {
        return 1;
    }
    if (totalPages <= 4) {
        return Math.min(2, cpu);
    }
    if (totalPages <= 16) {
        return Math.min(4, cpu);
    }
    if (totalPages <= 64) {
        return Math.min(8, cpu);
    }
    return Math.min(16, cpu);
}

4) 批量查询 SQL（offset 分页）

用于多生产者并发查询。

<select id="selectByOffset" resultType="com.liyao.learning.export.vo.TradeExportVO">
    SELECT
        id,
        trade_date AS tradeDate,
        trade_time AS tradeTime,
        buyer_id AS buyerId,
        amt
    FROM t_trade
    LIMIT #{offset}, #{pageSize}
</select>