Node.js Streams如何工作？

Question

我有一个关于Node.js流的问题 - 特别是它们在概念上如何工作.

关于如何使用流的文档不乏.但我很难找到数据流的工作方式.

我对网络通信HTTP的有限理解是,数据的完整"包"是来回发送的.与订购公司目录的个人类似,客户端向服务器发送GET(目录)请求,服务器响应目录.浏览器不会收到目录页面,而是整本书.

节点流可能是多部分消息吗？

我喜欢REST模型 - 特别是它是无状态的.浏览器和服务器之间的每次交互都是完全自包含且足够的.节点流因此不是RESTful吗？一位开发人员提到了与套接字管道的相似性,这使得连接保持打开状 回到我的目录订购示例,这将是一个电视购物广告,"但等等!还有更多!" 而不是完全包含的目录？

大部分流是接收器"下游"发送诸如"暂停"和"继续"上游之类的消息的能力.这些消息包括什么？他们是POST吗？

最后,我对Node如何工作的有限视觉理解包括这个事件循环.函数可以放在线程池的不同线程上,事件循环继续.但是不应该发送数据流保持事件循环被占用(即停止)直到流完成？它是如何继续监视下游的"暂停"请求？n事件循环是否将流放在池中的另一个线程上,当遇到"暂停"请求时,检索相关线程并暂停它？

我已经阅读了node.js文档,完成了nodeschool教程,构建了一个heroku应用程序,购买了两本书(真实的,自包含的,书籍,有点像以前说过的目录,可能不喜欢节点流),问了几个"节点"代码训练营的教练 - 所有人都谈论如何使用流,但没有人谈论下面实际发生的事情.

也许你遇到了一个很好的资源来解释这些是如何工作的？对于非CS心灵来说,也许是一个很好的拟人类比？

Answer 1

首先要注意的是:node.js流不仅限于HTTP请求.HTTP请求/网络资源只是node.js中流的一个示例.

Streams对于可以在小块中处理的所有内容都很有用.它们允许您以更小的块处理潜在的巨大资源,更容易适合您的RAM.

假设您有一个文件(大小为几千兆字节),并希望将所有小写转换为大写字符并将结果写入另一个文件.天真的方法将使用整个文件读取fs.readFile(为简洁省略了错误处理):

fs.readFile('my_huge_file', function (err, data) {
    var convertedData = data.toString().toUpperCase();

    fs.writeFile('my_converted_file', convertedData);
});

不幸的是,这个approch很容易压倒你的RAM,因为在处理它之前必须存储整个文件.您还会浪费宝贵的时间等待文件的读取.以较小的块处理文件是否有意义？在等待硬盘提供剩余数据时,您可以在获得第一个字节后立即开始处理:

var readStream = fs.createReadStream('my_huge_file');
var writeStream = fs.createWriteStream('my_converted_file');
readStream.on('data', function (chunk) {
    var convertedChunk = chunk.toString().toUpperCase();
    writeStream.write(convertedChunk);
});
readStream.on('end', function () {
    writeStream.end();
});

这种方法要好得多:

1. 您只需处理可轻松放入RAM的小部分数据.
2. 一旦第一个字节到达就开始处理,不要浪费时间做任何事情,而是等待.

打开流后,node.js将打开该文件并开始从中读取.一旦操作系统将一些字节传递给正在读取文件的线程,它将被传递给您的应用程序.

回到HTTP流:

1. 第一个问题在这里也是有效的.攻击者可能会向您发送大量数据以压倒您的RAM并删除(DoS)您的服务.
2. 然而,在这种情况下,第二个问题更为重要:网络可能非常慢(想想智能手机),并且可能需要很长时间才能由客户端发送所有内容.通过使用流,您可以开始处理请求并缩短响应时间.

暂停HTTP流:这不是在HTTP级别完成,而是更低.如果你暂停流node.js将停止从底层TCP套接字读取.然后发生的事情取决于内核.它仍然可以缓冲传入的数据,因此一旦完成当前工作就可以为您准备好了.它还可以通知TCP级别的发送方它应该暂停发送数据.应用程序不需要处理.这不关他们的事.事实上,发件人应用程序可能甚至没有意识到你不再积极阅读!

因此,它基本上是在数据可用时尽快提供,但不会压倒您的资源.底层的辛勤工作是由操作系统执行任何(例如net,fs,http),或者通过流的作家,你正在使用(例如,zlib它是一个Transform流,并且通常用螺栓固定fs或net).

Answer 2

下面的图表似乎是节点流类的非常准确的 10.000 英尺概览/图表。

它代表streams3，由Chris Dickinson贡献。

Answer 3

首先，什么是流？好吧，通过流，我们可以逐段处理意味着读取和写入数据，而无需完成整个读取或写入操作。因此，我们不必将所有数据保存在内存中来执行这些操作。

例如，当我们使用流读取文件时，我们读取部分数据，对其执行某些操作，然后释放内存，并重复此操作，直到处理整个文件。或者想想 YouTube 或 Netflix，它们都被称为流媒体公司，因为它们使用相同的原理流式传输视频。

因此，无需等到整个视频文件加载，而是逐段或分块进行处理，以便您甚至可以在下载整个文件之前就开始观看。所以这里的原理不仅仅是Node.JS。但对于整个计算机科学来说是普遍的。

正如您所看到的，这使得流成为处理大量数据的完美选择，例如视频或我们从外部源逐段接收的数据。此外，流式传输使数据处理在内存方面更加高效，因为无需将所有数据保留在内存中；在时间方面，因为我们可以在数据到达时开始处理数据，而不是等到所有数据都到达。

它们是如何在 Node.JS 中实现的：

因此，在 Node 中，有四种基本类型的流：可读流、可写流、双工流和转换流。但可读和可写是最重要的，可读流是我们可以读取和使用数据的流。流在核心 Node 模块中无处不在，例如，http 服务器收到请求时传入的数据实际上是可读的流。因此，随请求发送的所有数据都是一块一块的，而不是一大块。另外，文件系统的另一个例子是，我们可以使用 FS 模块的读取屏幕逐段读取文件，这对于大型文本文件实际上非常有用。

嗯，另一个需要注意的重要事情是流实际上是EventEmitter类的实例。这意味着所有流都可以发出并侦听命名事件。在可读流的情况下，它们可以发出，我们可以监听许多不同的事件。但最重要的两个是数据和结束事件。当有新的数据要使用时，会发出 data 事件；当没有更多的数据要使用时，会发出 end 事件。当然，我们可以对这些事件做出相应的反应。

最后，除了事件之外，我们还有可以在流上使用的重要函数。对于可读流来说，最重要的是管道和读取函数。超级重要的管道函数，它基本上允许我们将流插入在一起，将数据从一个流传递到另一个流，而不必担心事件。

接下来，可写流是我们可以写入数据的流。基本上，与可读流相反。一个很好的例子是我们可以发送回客户端的 http 响应，它实际上是一个可写流。这是一个我们可以写入数据的流。那么当我们想要发送数据时，我们就必须把它写到某个地方，对吗？某个地方是一个可写流，这很有意义，对吗？

例如，如果我们想向客户端发送一个大视频文件，我们会像 Netflix 或 YouTube 那样。现在说到事件，最重要的是排水事件和结束事件。最重要的函数是 write 和 end 函数。

关于双工流。它们只是同时可读和可写的流。这些不太常见。但无论如何，一个很好的例子是来自 net 模块的网络套接字。Web 套接字基本上只是客户端和服务器之间的通信通道，它可以双向工作，并且在建立连接后保持打开状态。

最后，转换流是双工流，因此流既可读又可写，同时可以在读取或写入数据时修改或转换数据。一个很好的例子是 zlib 核心模块来压缩数据，它实际上使用转换流。

*** 节点将这些 http 请求和响应实现为流，然后我们可以使用它们，我们可以使用每种类型的流可用的事件和函数来使用它们。我们当然也可以实现我们自己的流，然后使用这些相同的事件和函数来使用它们。

现在让我们尝试一些例子：

const fs = require('fs');
const server = require('http').createServer();
server.on('request', (req, res) =>{
    fs.readFile('./txt/long_file.txt', (err, data)=>{
        if(err) console.log(err);
        res.end(data);
    });
});
server.listen('8000','127.0.01', ()=>{
    console.log(this);
});

假设 long_file.txt 文件包含 1000000K 行，每行包含超过 100 个单词，所以这是一个包含大量数据的拥抱文件，现在在上面的示例中问题是通过使用readFile()函数节点将整个文件加载到内存中，因为只有将整个文件加载到内存节点后才能将数据传输为响应对象。

当文件很大，并且有大量请求到达您的服务器时，通过时间节点进程将很快耗尽资源，您的应用程序将停止工作，一切都会崩溃。

让我们尝试使用流来找到解决方案：

const fs = require('fs');
const server = require('http').createServer();

server.on('request', (req, res) =>{
    const readable = fs.createReadStream('./txt/long_file.txt');
    readable.on('data', chunk=>{
        res.write(chunk);
    });
    readable.on('end',()=>{
        res.end();
    })
    readable.on('error', err=>{
        console.log('err');
        res.statusCode=500;
        res.end('File not found');
    });
});

server.listen('8000','127.0.01', ()=>{
    console.log(this);
});

在上面的流示例中，我们有效地流式传输文件，我们正在读取文件的一部分，一旦可用，我们就使用响应流的 write 方法将其直接发送到客户端。然后，当下一个文件可用时，该文件将被发送，直到整个文件被读取并流式传输到客户端。

因此，流基本上完成了从文件中读取数据，将发出结束事件以表明不再有数据将写入此可写流。

通过上面的实践，我们解决了之前的问题，但是上面的例子仍然存在一个巨大的问题，称为背压。

问题在于，我们用来从磁盘读取文件的可读流比通过网络实际发送响应可写流的结果要快得多。这将压垮响应流，使其无法如此快地处理所有这些传入数据，这个问题称为背压。

解决方案是使用管道运算符，它将处理数据传入的速度和数据传出的速度。

const fs = require('fs');
const server = require('http').createServer();

server.on('request', (req, res) =>{
    const readable = fs.createReadStream('./txt/long_file.txt');
    readable.pipe(res);

});

server.listen('8000','127.0.01', ()=>{
    console.log(this);
});