零OOM导出:StreamingResponseBody原理&实战

叔牙

发布于 2025-08-02 12:56:41

3760

一、背景介绍

报表导出功能是基本上任何业务都需要的基本能力,稍微规模大一些的团队或者项目会用专门的BI工具实现,然而很多中小型企业和成本约束严格的项目,大多使用开源的Excel工具比如Apache POI以及阿里开源的EasyExcel进行导出,然而这些工具基本都是内存导出,那么在数据量比较大的情况下久容易造成内存溢出以及其他问题。

那么我们考虑一下,能否在不增加成本(不构建BI平台),不增加技术复杂度的情况下解决这种大数据量导出老大难的问题？

二、StreamingResponseBody是什么？

StreamingResponseBody是Spring框架从4.2版本增加的一个个用于处理异步响应的接口,特别适用于需要流式传输大文件或大量数据的场景。它允许开发者直接将数据写入HTTP响应的输出流,而无需将整个响应内容加载到内存中,尤其是在处理大文件下载或导出时,从而避免了内存溢出,并提高了程序性能。

@FunctionalInterface
public interface StreamingResponseBody {
    /**
	 * A callback for writing to the response body.
	 * @param outputStream the stream for the response body
	 * @throws IOException an exception while writing
	 */
    void writeTo(OutputStream outputStream) throws IOException;
}

1.异步处理

StreamingResponseBody配合Spring的异步处理机制,可以避免阻塞Servlet容器线程,提高服务器的并发处理能力。

2.流式传输

StreamingResponseBody允许数据以流的形式逐步发送给客户端,而不是等待整个文件生成完毕,这对于大文件下载和大数据导出非常重要。

3.低内存占用

由于数据是分块发送的,StreamingResponseBody可以显著降低服务器的内存占用,能够更好地利用系统资源,尤其是在高并发或者处理非常大的文件时,这一点对于java服务运行性能的稳定性是非常重要的。

4.提高用户体验

用户可以立即开始接收数据,而无需等待整个文件生成完毕,从而提升了用户体验。

5.应用场景

适用于大文件下载、大数据导出、实时数据流等场景。

三、基于StreamingResponseBody实现导出

由于StreamingResponseBody是spring框架web模块自带的能力,所以不用额外引入依赖,如果使用apache POI或者EasyExcel,需要引入对应的依赖,以EasyExcel为例。

1.添加依赖

<!--EasyExcel相关依赖-->
<dependency>
  <groupId>com.alibaba</groupId>
  <artifactId>easyexcel</artifactId>
</dependency>

2.添加导出实体类

按照EasyExcel用法编写导出数据实体类,字段使用@ExcelProperty注解标注自动生成表头。

@Data
public class OrderExportVO {
    @ExcelProperty("订单ID")
    private String orderId;
    @ExcelProperty("用户ID")
    private Long userId;
    @ExcelProperty("订单金额")
    private BigDecimal amount;
    @ExcelProperty("创建时间")
    @DateTimeFormat("yyyy-MM-dd HH:mm:ss")
    private Date createTime;
    @ExcelProperty("订单状态")
    private String status;
}

3.导出数据查询

编写模拟从数据库查询订单数据的服务。

@Service
@RequiredArgsConstructor
public class OrderExportService {
    private final OrderMapper orderMapper;
    // 分页查询订单数据
    public List<OrderExportVO> getOrderChunk(int page, int size) {
        PageHelper.startPage(page, size);
        return orderMapper.selectOrdersForExport();
    }
    // 获取订单总数
    public long getTotalOrders() {
        return orderMapper.countOrders();
    }
}

4.编写导出逻辑

编写导出数据的controller实现,将响应数据定义成ResponseEntity<StreamingResponseBody>或者StreamingResponseBody都可以,后续分析为什么这样做。

@RestController
@RequestMapping("/export")
@RequiredArgsConstructor
public class OrderExportController {
    private final OrderExportService orderExportService;
    private static final int PAGE_SIZE = 2000; // 每页大小
    @GetMapping("/orders")
    public ResponseEntity<StreamingResponseBody> exportOrders() {
        // 设置响应头
        HttpHeaders headers = new HttpHeaders();
        headers.setContentType(MediaType.APPLICATION_OCTET_STREAM);
        headers.setContentDisposition(
            ContentDisposition.builder("attachment")
            .filename("orders_" + System.currentTimeMillis() + ".xlsx")
            .build()
        );
        // 创建StreamingResponseBody
        StreamingResponseBody body = outputStream -> {
            try (ExcelWriter excelWriter = EasyExcel.write(outputStream, OrderExportVO.class).build()) {
                WriteSheet writeSheet = EasyExcel.writerSheet("订单数据").build();
                long total = orderExportService.getTotalOrders();
                int totalPages = (int) Math.ceil((double) total / PAGE_SIZE);
                // 分页写入
                for (int page = 1; page <= totalPages; page++) {
                    List<OrderExportVO> chunk = orderExportService.getOrderChunk(page, PAGE_SIZE);
                    excelWriter.write(chunk, writeSheet);
                    // 每页刷新输出流
                    outputStream.flush();
                    // 监控日志
                    if (page % 10 == 0) {
                        log.info("已导出 {} / {} 条数据", page * PAGE_SIZE, total);
                    }
                }
            } catch (Exception e) {
                log.error("导出失败", e);
                throw new RuntimeException("导出失败", e);
            }
        };
        return new ResponseEntity<>(body, headers, HttpStatus.OK);
    }
}

这样就可以实现订单或者其他报表数据大数据量导出的低内存占用、高性能实现了。

四、StreamingResponseBody工作原理

首先需要明确的是,StreamingResponseBody不是一种新的技术架构,使用的还是http协议,并且后端的逻辑处理也还是DispatcherServlet。那么它的工作原理是怎样子,接下来我们做一下分析。

先看一下DispatcherServlet核心逻辑doDispatch:

protected void doDispatch(HttpServletRequest request, HttpServletResponse response) throws Exception {
  HttpServletRequest processedRequest = request;
  HandlerExecutionChain mappedHandler = null;
  boolean multipartRequestParsed = false;
  WebAsyncManager asyncManager = WebAsyncUtils.getAsyncManager(request);
  try {
    ModelAndView mv = null;
    Exception dispatchException = null;
    try {
      //...省略...
      // Actually invoke the handler.
      mv = ha.handle(processedRequest, response, mappedHandler.getHandler());
      if (asyncManager.isConcurrentHandlingStarted()) {
        return;
      }
      //...省略...  
    }
    catch (Exception ex) {}
    processDispatchResult(processedRequest, response, mappedHandler, mv, dispatchException);
  }
  catch (Exception ex) {}
  catch (Throwable err) {}
  finally {
    if (asyncManager.isConcurrentHandlingStarted()) {
      if (mappedHandler != null) {
        mappedHandler.applyAfterConcurrentHandlingStarted(processedRequest, response);
      }
    }
  }
}

经过预处理和HandlerAdapter适配,HandlerAdapter#handle方法会真正调用controller中的接口方法,然后调用AbstractHandlerMethodAdapter的handle方法:

@Override
public final ModelAndView handle(HttpServletRequest request, HttpServletResponse response, Object handler)
    throws Exception {
  return handleInternal(request, response, (HandlerMethod) handler);
}

接着调用内部方法invokeHandlerMethod:

protected ModelAndView invokeHandlerMethod(HttpServletRequest request,
    HttpServletResponse response, HandlerMethod handlerMethod) throws Exception {
  ServletWebRequest webRequest = new ServletWebRequest(request, response);
  try {
    //...省略...  
    AsyncWebRequest asyncWebRequest = WebAsyncUtils.createAsyncWebRequest(request, response);
    asyncWebRequest.setTimeout(this.asyncRequestTimeout);
    WebAsyncManager asyncManager = WebAsyncUtils.getAsyncManager(request);
    asyncManager.setTaskExecutor(this.taskExecutor);
    asyncManager.setAsyncWebRequest(asyncWebRequest);
    asyncManager.registerCallableInterceptors(this.callableInterceptors);
    asyncManager.registerDeferredResultInterceptors(this.deferredResultInterceptors);
    if (asyncManager.hasConcurrentResult()) {
      Object result = asyncManager.getConcurrentResult();
      mavContainer = (ModelAndViewContainer) asyncManager.getConcurrentResultContext()[0];
      asyncManager.clearConcurrentResult();
      LogFormatUtils.traceDebug(logger, traceOn -> {
        String formatted = LogFormatUtils.formatValue(result, !traceOn);
        return "Resume with async result [" + formatted + "]";
      });
      invocableMethod = invocableMethod.wrapConcurrentResult(result);
    }
    invocableMethod.invokeAndHandle(webRequest, mavContainer);
    if (asyncManager.isConcurrentHandlingStarted()) {
      return null;
    }
    return getModelAndView(mavContainer, modelFactory, webRequest);
  }
  finally {
    webRequest.requestCompleted();
  }
}

组装请求和异步调用管理器后调用ServletInvocableHandlerMethod的invokeAndHandle方法:

public void invokeAndHandle(ServletWebRequest webRequest, ModelAndViewContainer mavContainer,
    Object... providedArgs) throws Exception {
  Object returnValue = invokeForRequest(webRequest, mavContainer, providedArgs);
  //...省略...  
  try {
    this.returnValueHandlers.handleReturnValue(
        returnValue, getReturnValueType(returnValue), mavContainer, webRequest);
  }
  catch (Exception ex) {
    throw ex;
  }
}

上述逻辑中invokeForRequest是执行真正的controller接口方法逻辑,执行完成后调用HandlerMethodReturnValueHandlerComposite的handleReturnValue方法处理返回值,它是一个复合返回值处理器,封装了返回值处理列表支持多种返回值处理:

public void handleReturnValue(@Nullable Object returnValue, MethodParameter returnType,
    ModelAndViewContainer mavContainer, NativeWebRequest webRequest) throws Exception {
  HandlerMethodReturnValueHandler handler = selectHandler(returnValue, returnType);
  if (handler == null) {
    throw new IllegalArgumentException("Unknown return value type: " + returnType.getParameterType().getName());
  }
  handler.handleReturnValue(returnValue, returnType, mavContainer, webRequest);
}

这里会先调用selectHandler方法选择合适的返回值处理器:

private HandlerMethodReturnValueHandler selectHandler(@Nullable Object value, MethodParameter returnType) {
  boolean isAsyncValue = isAsyncReturnValue(value, returnType);
  for (HandlerMethodReturnValueHandler handler : this.returnValueHandlers) {
    if (isAsyncValue && !(handler instanceof AsyncHandlerMethodReturnValueHandler)) {
      continue;
    }
    if (handler.supportsReturnType(returnType)) {
      return handler;
    }
  }
  return null;
}

而前边我们编写的接口返回值类型是ResponseEntity<StreamingResponseBody>类型,所以可以找到StreamingResponseBodyReturnValueHandler,可以看到她所支持的返回值类型是StreamingResponseBody或者ResponseEntity<StreamingResponseBody>,也就是前边我们所说的使用StreamingResponseBody要把接口返回值定义成StreamingResponseBody或者ResponseEntity的原因。

选择好合适的返回值处理器之后,会调用其handleReturnValue方法,对于StreamingResponseBody类型则会调用StreamingResponseBodyReturnValueHandler的handleReturnValue方法:

public void handleReturnValue(@Nullable Object returnValue, MethodParameter returnType,
    ModelAndViewContainer mavContainer, NativeWebRequest webRequest) throws Exception {
  //...省略...
  HttpServletResponse response = webRequest.getNativeResponse(HttpServletResponse.class);
  ServerHttpResponse outputMessage = new ServletServerHttpResponse(response);
  if (returnValue instanceof ResponseEntity) {
    ResponseEntity<?> responseEntity = (ResponseEntity<?>) returnValue;
    response.setStatus(responseEntity.getStatusCodeValue());
    outputMessage.getHeaders().putAll(responseEntity.getHeaders());
    returnValue = responseEntity.getBody();
    //...省略...
  }
  ServletRequest request = webRequest.getNativeRequest(ServletRequest.class);
  ShallowEtagHeaderFilter.disableContentCaching(request);
  StreamingResponseBody streamingBody = (StreamingResponseBody) returnValue;
  Callable<Void> callable = new StreamingResponseBodyTask(outputMessage.getBody(), streamingBody);
  WebAsyncUtils.getAsyncManager(webRequest).startCallableProcessing(callable, mavContainer);
}

上述handleReturnValue方法的核心逻辑是创建StreamingResponseBodyTask任务,然后使用异步操作管理器调用异步线程池完成StreamingResponseBody数据写入到HttpServletResponse响应数据流中。

private static class StreamingResponseBodyTask implements Callable<Void> {
  private final OutputStream outputStream;
  private final StreamingResponseBody streamingBody;
  public StreamingResponseBodyTask(OutputStream outputStream, StreamingResponseBody streamingBody) {
    this.outputStream = outputStream;
    this.streamingBody = streamingBody;
  }
  @Override
  public Void call() throws Exception {
    this.streamingBody.writeTo(this.outputStream);
    this.outputStream.flush();
    return null;
  }
}

StreamingResponseBodyTask的实现很简单,调用StreamingResponseBody的writeTo方法将数据写入到HttpServletResponse的输出流。

等数据写入完成后由DispatcherServlet调用调用异步操作管理器完成异步写入的收尾工作。对于使用StreamingResponseBody实现数据导出的大致工作流程如下:

StreamingResponseBody对于流式数据响应的工作原理时序图如下:

五、总结

对于大数据量报表导出场景,使用StreamingResponseBody与传统的内存导出方案对比如下:

总结来说StreamingResponseBody对于响应数据的处理是一种“化整为零”的做法,将完整的数据块拆成零散的小块进行数据传输,从而降低了服务端的内存压力、数据传输带宽压力等。当然它的价值远不止单纯的数据报表导出这么单一,从业务场景来看StreamingResponseBody可以在以下若干场景发挥巨大的作用:

大数据量文件导出: 本篇着重分析介绍的场景,大数据量报表导出
实时日志流输出：日志系统需实时展示或导出实时产生的日志（如服务器错误日志、用户操作日志）,传统方式需等待日志写入文件后再下载,无法满足“边生成边下载”的实时需求,StreamingResponseBody可以完美解决这个问题
大文件分片下载: 超大型文件(如GB级视频、备份包)直接下载时,浏览器或客户端可能因超时、内存限制无法完整接收,并且极容易造成服务端OOM,以及单次下载失败需重新开始。使用StreamingResponseBody可以将大文件拆分为多个分片(如每 10MB一个分片),逐片流式写入响应流,客户端可分段下载并合并
API流式响应&股票行情: 传统API响应需等待所有数据处理完成后返回,无法满足“边处理边返回”的实时需求(如数据统计、批量计算结果),特别是股票行情等实时更新的数据，传统方案往往需要轮询获取数据,带来额外的网络开销。使用 StreamingResponseBody,可以建立长连接,让服务器主动推送最新数据
低带宽高效传输: 在网络带宽有限或者弱网环境(如偏远地区、基站不完善地方),大文件的完整传输易受网络波动影响,导致数据传输失败或耗时过长。使用StreamingResponseBody调整响应数据块大小能很大程度上解决或者改善弱网数据传输问题。

本文参与腾讯云自媒体同步曝光计划，分享自微信公众号。

原始发表：2025-07-29，如有侵权请联系 cloudcommunity@tencent.com 删除

amp

本文分享自 PersistentCoder 微信公众号，前往查看

如有侵权，请联系 cloudcommunity@tencent.com 删除。

本文参与腾讯云自媒体同步曝光计划，欢迎热爱写作的你一起参与！

登录后参与评论

0 条评论

热度