端到端反应式流式处理 RESTful 服务

Question

我已经阅读了this post，但我不知道这是不是一个愚蠢的问题。我是响应式编程的新手。

问题很简单：假设我有一个完全反应式的后端，我怎样才能流式传输到浏览器（例如大文本），并在每个块来自服务器时立即将它们打印给用户？

可能我遗漏了一些重要的概念点，但我需要知道的是：我可以发送一小部分数据（从服务器到浏览器），用一个 HTTP GET（？或不是）请求？关键是：我可以写那些小部分，直到整个数据都发送出去？

感谢您的帮助。我之前确实尝试过google，但我总是得到关于其他概念的信息（如websockets、long-polling、react.js），我认为情况并非如此。

编辑：我不是要求特定的 API ou 库。我只是想理解这个概念，例如：“You CANNOT do this with HTTP protocol，句号！”或“这与反应式编程无关，您对流的真正含义感到困惑。请参阅'something_else.js'”。

EDIT2：我用这种方法使用 spring-mvc (spring-boot) 做了一个小型休息控制器：

@RequestMapping(value = "/flush", method = RequestMethod.GET)
public void flushTest(HttpServletResponse response) {

    response.setContentType("text/html");
    response.setStatus(SC_OK);

    try (ServletOutputStream outputStream = response.getOutputStream()) {
        for (int i = 0; i < 3; i++) {
            String chunk = "chunk_" + i;
            outputStream.print(chunk);
            outputStream.flush();
            response.flushBuffer();
        }
    } catch (IOException e) {
        e.printStackTrace();
    }

}

在浏览器中，即使使用了flushBuffer，也只有整个响应到达。经过一番研究，这可能是 TOMCAT 问题，我更改了 pom.xml，包括 undertow：

<dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-undertow</artifactId>
</dependency>

使用默认配置：

server.undertow.accesslog.enabled=true
server.undertow.accesslog.dir=target/logs
server.undertow.accesslog.pattern=combined
server.compression.enabled=true
server.compression.min-response-size=1l

此时，在我的 IDE 中使用调试，每个块都在 flushBuffer 之后写入。

所以，我相信这可能是一个 TOMCAT 配置。长话短说，我知道在“流式解决方案”方面有很多事情可以帮助我，感谢所有 cmets，但我的问题有点概念化。

Answer 1

在 HTTP 之上有一个协议，它允许使用 HTTP 传输将服务器响应以较小的、准备好使用的块传送到浏览器。它被称为 SSE（服务器发送事件）或 EventSource，here 是关于该主题的非常完整的文章。

还有其他方法可以使用 HTTP 协议流式传输数据。一种这样的替代方案是 JSON 流，您可以在其中从服务器在线写入部分响应，并在 JSON 块到达时使用它们。在消费端，一个流行的库称为Oboe.js，在服务器端，您基本上需要在响应线路上写入部分数据，当您想要发送它时，或者当它可用时。

对于这两种方法，Rx 对于处理服务器端的流逻辑很有用。您将这些片段建模为流（处理错误等），最终，在订阅者中，您将单个发射写入网络，并在 Observable 完成时关闭响应。

在客户端，EventSource 事件可以包装在一个新的 Observable 中（通过Rx.Observable.create()）并作为流进一步处理。 Oboe.js 事件也可以转换为 Observables（通过Rx.Observable.fromEvent()。