从 Spring REST 控制器返回流

Question

如果有可能从 Spring RestController 返回 Stream，我很好奇

@RestController
public class X {
  @RequestMapping(...)
  public Stream<?> getAll() { ... }
}

可以做这样的事情吗？我试过了，Spring 返回的不是流的值。

我应该继续返回List<?>吗？

Answer 1

这也可以通过 Spring MVC Controller 来完成，但是有几个问题：Spring Data JPA Repository 的限制，数据库是否支持 Holdable Cursors（ResultSet Holdability）以及 Jackson 的版本。

我难以理解的关键概念是，Java 8 Stream 返回一系列函数，这些函数在终端操作中执行，因此数据库必须在执行终端操作。

Spring Data JPA 限制

我发现 Spring Data JPA 文档没有为 Java 8 Streams 提供足够的详细信息。看起来您可以简单地声明 Stream<MyObject> readAll()，但我需要用 @Query 注释该方法以使其工作。我也无法使用 JPA 标准 API Specification。所以我不得不接受一个硬编码的查询，比如：

@Query("select mo from MyObject mo where mo.foo.id in :fooIds")
Stream<MyObject> readAllByFooIn(@Param("fooIds") Long[] fooIds);

可保持光标

如果您有一个支持可保持游标的数据库，则在提交事务后可以访问结果集。这一点很重要，因为我们通常用@Transactional 注释我们的@Service 类方法，因此如果您的数据库支持可保持游标，则可以在服务方法返回后访问ResultSet，即在@Controller 方法中。如果数据库不支持可保持游标，例如MySQL，您需要将@Transaction 注解添加到控制器的@RequestMapping 方法中。

所以现在 ResultSet 可以在 @Service 方法之外访问，对吧？这又取决于可持性。对于 MySQL，它只能在 @Transactional 方法中访问，因此以下将起作用（尽管违背了使用 Java 8 Streams 的全部目的）：

@Transaction @RequestMapping(...)
public List<MyObject> getAll() {
   try(Stream<MyObject> stream = service.streamAll) {
        return stream.collect(Collectors.toList())
    };
}

但不是

@Transaction @RequestMapping
public Stream<MyObject> getAll() {
    return service.streamAll;
}

因为 终端操作符 在您的@Controller 中不是，它会在控制器方法返回后在 Spring 中发生。

在不支持可保持游标的情况下将流序列化为 JSON

要将流序列化为没有可保持游标的 JSON，请将 HttpServletResponse response 添加到控制器方法中，获取输出流并使用 ObjectMapper 写入流。使用 FasterXML 3.x，您可以调用 ObjectMapper().writeValue(writer, stream)，但使用 2.8.x，您必须使用流的迭代器：

@RequestMapping(...)
@Transactional
public void getAll(HttpServletResponse response) throws IOException {
    try(final Stream<MyObject> stream = service.streamAll()) {
        final Writer writer = new BufferedWriter(new OutputStreamWriter(response.getOutputStream()));
        new ObjectMapper().writerFor(Iterator.class).writeValue(writer, stream.iterator());
    }
}

后续步骤

我接下来的步骤是尝试在 Callable WebAsyncTask 中重构它，并将 JSON 序列化移动到服务中。

参考文献

• 请务必阅读 Marko Topolnik 的博客文章 https://www.airpair.com/java/posts/spring-streams-memory-efficiency，否则我不知道从哪里开始。
• MySQL >5.0.2 现在支持游标，所以你可以在连接字符串中添加useCursorFetch=true -- https://dev.mysql.com/doc/connector-j/5.1/en/connector-j-reference-configuration-properties.html
• Java 8 Stream 的 FasterXml 序列化 -- https://stackoverflow.com/a/37979665/2562746

Answer 2

您可以在 Spring 5.0 / WebFlux 中流式传输实体。

看看这个例子 REACTIVE Rest Controller (spring.main.web-application-type: "REACTIVE")：

@RestController
public class XService {

    class XDto{
        final int x;
        public XDto(int x) {this.x = x;}
    }

    Stream<XDto> produceX(){
        return IntStream.range(1,10).mapToObj(i -> {
            System.out.println("produce "+i);
            try {Thread.sleep(1000);} catch (InterruptedException e) {e.printStackTrace();}
            return new XDto(i);
        });
    }

    // stream of Server-Sent Events (SSE)
    @GetMapping(value = "/api/x/sse", 
    produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public Flux<XDto> getXSse() {
        return Flux.fromStream(produceX());
    }

    // stream of JSON lines
    @GetMapping(value = "/api/x/json-stream", 
    produces = MediaType.APPLICATION_STREAM_JSON_VALUE)
    public Flux<XDto> getAllJsonStream() {
        return Flux.fromStream(produceX());
    }

    // same as List<XDto> - blocking JSON list
    @GetMapping(value = "/api/x/json-list", 
    produces = MediaType.APPLICATION_JSON_VALUE)
    public Flux<XDto> getAll() {
        return Flux.fromStream(produceX());
    }
}

Spring Framework 5.0 - WebFlux:

Spring 5.0 中的新反应式堆栈 Web 框架是完全反应式和非阻塞的。适用于线程数较少的事件循环式处理。

Server-Sent Events (SSE):

服务器发送事件是一种标准，描述了一旦建立了初始客户端连接，服务器如何向客户端发起数据传输。

WebSockets vs. Server-Sent events/EventSource