流式批量下载 + SSE实时进度返回

本文档基于当前导出实现，说明“流式批量下载 + SSE 实时进度回传”的能力覆盖、设计说明与核心代码。

能力覆盖

✅ Keyset Pagination（稳定分页）
✅ 多条件排序
✅ 大数据量导出（50W+）
✅ 流式写 Excel（EasyExcel）
✅ 实时导出进度回传
✅ 顺序 100% 与页面一致
❌ 不使用 offset
❌ 不存在翻页错乱

设计说明

1) Keyset Pagination（稳定分页）

使用“基于游标的分页”而非 offset。每次查询携带上一次的最后一条记录，确保在大数据量下分页稳定、性能线性、不会因为中途插入导致错页。

游标字段：

• trade_date、trade_time、id

排序与游标约束一致：

• ORDER BY trade_date DESC, trade_time DESC, id DESC
• WHERE (trade_date < ? OR (trade_date = ? AND trade_time < ?) OR (trade_date = ? AND trade_time = ? AND id < ?))

2) 多条件排序 + 顺序一致

分页顺序严格与页面一致，因为：

• 导出与查询使用同一组排序字段
• 下一页永远从上一页最后一条记录向后推进

3) 大数据量导出（50W+）

采用分页拉取 + 流式写入，不在内存中持有全量数据，内存占用稳定。

4) 流式写 Excel（EasyExcel）

通过 EasyExcel.write(...).build() 创建 ExcelWriter，每批数据写入同一个 WriteSheet，导出过程中输出流持续写入。

5) SSE 实时进度回传

导出过程中会不断回传 ExportProgress，前端订阅 SSE 即可实时看到进度。

6) 不使用 offset / 不存在翻页错乱

所有分页均基于游标推进，避免 offset 在大数据量场景下的性能下降与数据错页问题。

核心代码实现

1) Keyset 分页 SQL

文件：src/main/resources/mapper/TradeMapper.xml

<select id="selectByKeyset" 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
    <where>
        <if test="cursor != null">
            AND (
                trade_date &lt; #{cursor.tradeDate}
                OR (trade_date = #{cursor.tradeDate}
                    AND trade_time &lt; #{cursor.tradeTime})
                OR (trade_date = #{cursor.tradeDate}
                    AND trade_time = #{cursor.tradeTime}
                    AND id &lt; #{cursor.id})
            )
        </if>
    </where>
    ORDER BY
        trade_date DESC,
        trade_time DESC,
        id DESC
    LIMIT #{pageSize}
</select>

2) 导出服务（分页 + 流式写入 + 进度回调）

文件：src/main/java/com/liyao/learning/export/service/TradeExportService.java

public void export(
        OutputStream out,
        ExportConfig config,
        Consumer<ExportProgress> progressCallback
) {
    long total = tradeMapper.countForExport();
    ExportProgress progress = new ExportProgress(total);

    TradeCursor cursor = null;
    ExcelWriter writer = EasyExcel.write(out, TradeExportVO.class).build();

    int sheetNo = 0;
    int sheetRowCount = 0;
    WriteSheet sheet = EasyExcel.writerSheet(sheetNo, "Sheet1").build();

    try {
        while (true) {
            List<TradeExportVO> page =
                    tradeMapper.selectByKeyset(cursor, config.getPageSize());

            if (page == null || page.isEmpty()) {
                break;
            }

            if (config.isSplitSheet()
                    && sheetRowCount + page.size() > config.getSheetSize()) {
                sheetNo++;
                sheetRowCount = 0;
                sheet = EasyExcel.writerSheet(
                        sheetNo, "Sheet" + (sheetNo + 1)).build();
            }

            writer.write(page, sheet);
            sheetRowCount += page.size();

            progress.add(page.size());
            progressCallback.accept(progress);

            TradeExportVO last = page.get(page.size() - 1);
            cursor = new TradeCursor();
            cursor.setTradeDate(last.getTradeDate());
            cursor.setTradeTime(last.getTradeTime());
            cursor.setId(last.getId());
        }
    } finally {
        writer.finish();
    }
}

3) SSE 进度订阅 + 文件下载

文件：src/main/java/com/liyao/learning/export/controller/TradeExportController.java

SSE 进度通道：

@GetMapping(value = "/export/progress", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public SseEmitter exportProgress(
        @RequestParam(required = false) String exportId
) {
    String resolvedExportId = exportId == null || exportId.isBlank()
            ? UUID.randomUUID().toString()
            : exportId;
    SseEmitter emitter = new SseEmitter(0L);
    List<SseEmitter> emitters = progressEmitters.computeIfAbsent(
            resolvedExportId, key -> new CopyOnWriteArrayList<>());
    emitters.add(emitter);
    emitter.onCompletion(() -> removeEmitter(resolvedExportId, emitter));
    emitter.onTimeout(() -> removeEmitter(resolvedExportId, emitter));
    emitter.onError((ex) -> removeEmitter(resolvedExportId, emitter));
    sendEvent(emitter, "exportId", resolvedExportId);
    return emitter;
}

下载接口（流式输出 + 进度回传 + 完成关闭 SSE）：

@GetMapping(value = "/export")
public ResponseEntity<StreamingResponseBody> export(
        @RequestParam(required = false) String exportId
) {
    String resolvedExportId = exportId == null || exportId.isBlank()
            ? UUID.randomUUID().toString()
            : exportId;
    ExportConfig config = new ExportConfig();
    String filename = "trade_" + LocalDateTime.now().format(FILE_TS) + ".xlsx";

    StreamingResponseBody body = outputStream -> tradeExportService.export(
            outputStream,
            config,
            progress -> broadcastProgress(resolvedExportId, progress)
    );
    StreamingResponseBody wrappedBody = outputStream -> {
        try {
            body.writeTo(outputStream);
            completeExport(resolvedExportId);
        } catch (Exception ex) {
            errorExport(resolvedExportId, ex);
            throw ex;
        }
    };

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

进度事件：

private void broadcastProgress(String exportId, ExportProgress progress) {
    List<SseEmitter> emitters = progressEmitters.get(exportId);
    if (emitters == null || emitters.isEmpty()) {
        return;
    }
    for (SseEmitter emitter : emitters) {
        sendEvent(emitter, "progress", progress);
    }
}

完成/异常事件：

private void completeExport(String exportId) {
    List<SseEmitter> emitters = progressEmitters.get(exportId);
    if (emitters == null || emitters.isEmpty()) {
        return;
    }
    for (SseEmitter emitter : emitters) {
        sendEvent(emitter, "completed", "ok");
        emitter.complete();
    }
    progressEmitters.remove(exportId);
}

private void errorExport(String exportId, Exception ex) {
    List<SseEmitter> emitters = progressEmitters.get(exportId);
    if (emitters == null || emitters.isEmpty()) {
        return;
    }
    for (SseEmitter emitter : emitters) {
        sendEvent(emitter, "error", ex.getMessage());
        emitter.completeWithError(ex);
    }
    progressEmitters.remove(exportId);
}

使用流程

1. 打开进度通道
   GET /trade/export/progress
   收到 event: exportId 后拿到 exportId

2. 发起导出下载
   GET /trade/export?exportId=...

3. SSE 收到 progress 事件，完成时收到 completed 事件并断开