使用 Node.js 流式传输数据答案

【问题标题】：Streaming data with Node.js使用 Node.js 流式传输数据
【发布时间】：2011-02-03 06:08:18
【问题描述】：

我想知道是否可以使用 Node.js 将数据从服务器流式传输到客户端。我想向 Node.js 发布一个 AJAX 请求，然后保持连接打开并不断将数据流式传输到客户端。客户端会收到这个流并不断更新页面。

更新：

作为this answer 的更新 - 我无法让它工作。在您致电close 之前，不会发送response.write。我已经设置了一个示例程序，用于实现此目的：

Node.js：

var sys = require('sys'), 
http = require('http');
http.createServer(function (req, res) {
    res.writeHead(200, {'Content-Type': 'text/html'});
    var currentTime = new Date();
    setInterval(function(){
        res.write(
            currentTime.getHours()
            + ':' + 
            currentTime.getMinutes()
            + ':' +
            currentTime.getSeconds()
        );
    },1000);
}).listen(8000);

HTML：

<html>
    <head>
        <title>Testnode</title>
    </head>

    <body>
        <!-- This fields needs to be updated -->
        Server time: <span id="time">&nbsp;</span>

        <!-- import jQuery from google -->
        <script type="text/javascript" src="http://ajax.googleapis.com/ajax/libs/jquery/1.4.2/jquery.min.js"></script>

        <!-- import jQuery -->
        <script type="text/javascript">
            $(document).ready(function(){
            // I call here node.localhost nginx ports this to port 8000
                $('#time').load('http://node.localhost');
            });
        </script>
    </body>
</html>

使用这种方法，在我致电close() 之前，我不会得到任何回报。这是可能的吗，还是我应该采用长轮询方法，而不是在一个进来时再次调用加载函数？

【问题讨论】：

另外我不认为这是流媒体，这只是将内容分块。是的，节点会发送部分答案，但每个 html 服务器都会这样做。您从节点获得的流媒体优势指的是其他东西。
另外，在您的示例中，您没有 pause、resume、flush 方法。（读取）流是为您提供数据的东西，尽可能快（或尽可能快）。你告诉它暂停，然后它就暂停了。你告诉它继续等等。

标签： jquery html ajax node.js data-stream

【解决方案1】：

这是可能的。只需多次使用response.write()。

var body = ["hello world", "early morning", "richard stallman", "chunky bacon"];
// send headers
response.writeHead(200, {
  "Content-Type": "text/plain"
});

// send data in chunks
for (piece in body) {
    response.write(body[piece], "ascii");
}

// close connection
response.end();

您可能必须每 30 秒左右关闭并重新打开连接。

编辑：这是我实际测试的代码：

var sys = require('sys'),
http = require('http');
http.createServer(function (req, res) {
    res.writeHead(200, {'Content-Type': 'text/html'});
    var currentTime = new Date();
    sys.puts('Starting sending time');
    setInterval(function(){
        res.write(
            currentTime.getHours()
            + ':' +
            currentTime.getMinutes()
            + ':' +
            currentTime.getSeconds() + "\n"
        );

        setTimeout(function() {
            res.end();
        }, 10000);

    },1000);
}).listen(8090, '192.168.175.128');

我通过 Telnet 连接到它，它确实给出了分块响应。但是要在 AJAX 浏览器中使用它必须支持 XHR.readyState = 3（部分响应）。据我所知，并非所有浏览器都支持这一点。所以你最好使用长轮询（或 Chrome/Firefox 的 Websockets）。

EDIT2：另外，如果您使用 nginx 作为 Node 的反向代理，它有时会希望收集所有块并立即将其发送给用户。你需要调整它。

【讨论】：

我正在测试这种方法，但在我的页面上通过 Ajax 访问 node.js 脚本时遇到问题。我认为这与一些跨域的事情有关。稍后我将对此进行处理，看看是否按预期工作。
jquery.load 在触发回调之前等待所有主体到达。你需要使用不同的东西。见api.jquery.com/load
我会在我的节点机器上检查它（0.1.31）
是的，也许 longpoll 会更好。
要使这个示例工作：移动“var currentTime = new Date();”进入 setinterval 函数。

【解决方案2】：

看看 Sockets.io。它提供 HTTP/HTTPS 流，并使用各种传输方式：

WebSocket
WebSocket over Flash（+ XML 安全策略支持）
XHR 轮询
XHR 多部分流
永远的 iframe
JSONP 轮询（用于跨域）

还有！它与 Node.JS 无缝协作。它也是一个 NPM 包。

https://github.com/LearnBoost/Socket.IO

https://github.com/LearnBoost/Socket.IO-node

【讨论】：

Socket.IO 对此效率低下，因为他正在寻找从服务器到客户端的单向流。 SSE 非常适合。
这个解决方案对我帮助很大。我想将网络摄像头视频 + 麦克风音频从客户端流式传输到服务器。我该怎么做？ stackoverflow.com/questions/27712381/…

【解决方案3】：

你也可以中止无限循环：

app.get('/sse/events', function(req, res) {
    res.header('Content-Type', 'text/event-stream');

    var interval_id = setInterval(function() {
        res.write("some data");
    }, 50);

    req.socket.on('close', function() {
        clearInterval(interval_id);
    }); 
});

这是 expressjs 的一个例子。我相信没有 expressjs 会是这样的。

【讨论】：

【解决方案4】：

这现在可以通过 Node.js、pg-query-stream 和 Materialize 实现。

Materialize 与 PostgreSQL 兼容，这意味着 Node.js 应用程序可以使用任何现有的 PostgreSQL 客户端与 Materialize 进行交互。

但与 PostgreSQL 不同的是，通过 Materialize，您可以利用 Node.js 应用程序中增量更新的物化视图，而不是在某个时间点查询 Materialize 以获取视图的状态，而是使用 TAIL 语句来请求视图变化时的更新流。

示例应用如下所示：

import express from 'express'
import pg from 'pg'
import QStream from 'pg-query-stream'

const app = express();
const port = 3000

app.get('/questions', async (request, response) => {

    const client = new pg.Client('postgres://materialize@SERVER_IP:6875/materialize');

    await client.connect();

    const query = new QStream('TAIL your_materialized_view WITH (PROGRESS)', [], {batchSize: 1});

    const stream = client.query(query);

    response.setHeader('Content-Type',  'text/event-stream');

    for await (const event of stream) {
        if(event.id){
            response.write(`data: ${JSON.stringify(event)}\n`);
        }
    }

})

app.listen(port)

资源：

【讨论】：