java - Spring 将 JPA 流式传输到 HTTP

Question

我想将大数据（CSV 导出）从 JPA 流式传输到 HTTP 响应。在对 SO 和几个教程进行了多次讨论之后，我得出了以下代码，但是当我运行代码时，我得到一个错误org.hibernate.exception.GenericJDBCException: could not advance using next()并且SQL Error: 90007 The object is already closed [90007-197]. 不管我是否放在StreamingResponseBody stream = ...try/catch 块之外。我怀疑 DB 连接在调用 StreamResponseBody 回调时及时关闭。如何正确构建代码？

public class AdminExportController extends AbstractApiController {
  ...
  @GetMapping(path = "/export/{publicId}")
  @ApiOperation("Download file containing export data stored in the database")
  @Secured(PERM_EXPORTS_DOWNLOAD)
  public ResponseEntity<StreamingResponseBody> streamDataExportFile(@PathVariable("publicId") String publicId, final HttpServletResponse response) {
    DataExport dataExport = dataExportService.getFile(publicId);
    String filename = dataExport.getType().name() + "_" + LocalDate.now().toString() + ".csv";

    HttpHeaders headers = new HttpHeaders();
    headers.setContentType(ZenMediaType.TEXT_CSV);
    headers.setContentDispositionFormData(filename, filename);
    headers.setCacheControl("must-revalidate, post-check=0, pre-check=0");

    StreamingResponseBody stream = null;
    try(Stream<String> lines = dataExportService.streamFileContent(dataExport.getType())) {
      PrintWriter printer = response.getWriter();
      printer.println(dataExport.getCsvHeader());
      stream = out -> lines.forEach(line -> {
        printer.println(line);
      });
      printer.flush();
    } catch(IOException e) {
      e.printStackTrace();
    }
    return new ResponseEntity<>(stream, headers, HttpStatus.OK);
  }
...
}

@Service
public class DataExportServiceImpl extends AbstractGwServiceImpl implements DataExportService {
...
  @Override
  @Transactional(readOnly = true)
  public Stream<String> streamFileContent(DataExport.DataExportType type) {
    return dataExportLineRepository.streamAllByType(type).map(dataExportLine -> {
      String content = dataExportLine.getContent();
      entityManager.detach(dataExportLine);
      return content;
    });
  }
...
}

public interface DataExportLineRepository extends MyJpaRepository<DataExportLine, Long> {
  @QueryHints(value = {
     @QueryHint(name = HINT_FETCH_SIZE, value = "0"),
     @QueryHint(name = HINT_CACHEABLE, value = "false"),
     @QueryHint(name = READ_ONLY, value = "true")
  })
  @Query("select del from DataExportLine as del where del.type = :type")
  Stream<DataExportLine> streamAllByType(@Param("type") DataExport.DataExportType type);
}

标签： javaspringspring-data-jpajava-stream

Answer 1

在对 SO 进行了更多挖掘之后，我通过直接写入 HTTP 响应来找到这个解决方案。但我仍然很好奇它是否可以与 StreamingResponseBody 一起使用。

public class AdminExportController extends AbstractApiController {
  ...
  @GetMapping(path = VERSION_1 + "/{publicId}")
  @ApiOperation("Download file containing export data stored in the database")
  @Secured(PERM_EXPORTS_DOWNLOAD)
  public void writeDataExportFile(@PathVariable("publicId") String publicId, final HttpServletResponse response) {
    DataExport dataExport = dataExportService.getFile(publicId);
    String filename = dataExport.getType().name() + "_" + LocalDate.now().toString() + ".csv";

    response.setContentType(ZenMediaType.TEXT_CSV_VALUE);
    response.setHeader("Content-Disposition", "attachment; filename=\"" + filename + "\"");
    response.setHeader("Cache-Control", "must-revalidate, post-check=0, pre-check=0");

    try {
      PrintWriter writer = response.getWriter();
      writer.println(dataExport.getCsvHeader());
      writer.flush();
      dataExportService.writeFileContent(writer, dataExport.getType());
    } catch(IOException e) {
      e.printStackTrace();
    }
  }
...
}

@Service
public class DataExportServiceImpl extends AbstractGwServiceImpl implements DataExportService {
...
  @Override
  public void writeFileContent(PrintWriter writer, DataExport.DataExportType type) {
    dataExportLineRepository.streamAllByType(type).forEach(dataExportLine -> {
      String content = dataExportLine.getContent();
      entityManager.detach(dataExportLine);
      writer.println(content);
      writer.flush();
    });
  }
...
}

public interface DataExportLineRepository extends MyJpaRepository<DataExportLine, Long> {
  @QueryHints(value = {
     @QueryHint(name = HINT_FETCH_SIZE, value = "0"),
     @QueryHint(name = HINT_CACHEABLE, value = "false"),
     @QueryHint(name = READ_ONLY, value = "true")
  })
  @Query("select del from DataExportLine as del where del.type = :type")
  Stream<DataExportLine> streamAllByType(@Param("type") DataExport.DataExportType type);
}