【第2章 Node.js基础】2.7 Node.js 的流(一) 可读流

🌈 Node.js 的流

🚀什么是流

流不是 Node.js 特有的概念。它们是几十年前在 Unix 操作系统中引入的。

我们可以把流看作这些数据的集合，就像液体一样，我们先把这些液体保存在一个容器里（流的内部缓冲区 BufferList），等到相应的事件触发的时候，我们再把里面的液体倒进管道里，并通知其他人在管道的另一侧拿自己的容器来接里面的液体进行处理。

Node.js中的流是一种处理数据的抽象接口，它提供了一种有效的方式（按需处理数据）来读取或写入大量数据，而无需一次性将整个数据加载到内存中。流可以被视为数据在某段时间内从一个点移动到另一个点的序列，类似于水从一根管子的一端流到另一端。

🚀Node.js流的好处

Node.js流解决了处理大量数据时可能遇到的内存消耗和性能问题。以下是使用Node.js流的好处：

1. 内存效率：使用流可以按需处理数据，而不需要一次性将整个数据加载到内存中。这意味着即使处理大量数据，也可以节省大量的内存空间。流以块的形式处理数据，每次只加载和处理一小部分数据，然后将其释放，从而减少了内存的使用量。

2. 性能优化：由于流以块的形式处理数据，可以实现并行处理，提高处理速度。流还可以通过流水线（pipeline）的方式将多个操作连接起来，实现数据的连续处理，从而提高整体性能。

3. 响应性：使用流可以实现实时处理数据，即在数据到达时立即进行处理，而不需要等待整个数据加载完成。这对于需要实时处理数据的应用程序非常重要，例如实时日志分析、实时聊天等。

4. 可扩展性：流是可组合的，可以将多个流连接起来形成更复杂的数据处理流程。这种可组合性使得代码更加模块化和可重用，便于扩展和维护。

5. 适用于各种场景：Node.js流可以用于处理各种类型的数据，包括文件、网络请求、数据库查询结果等。无论是处理大文件、处理网络请求还是处理数据库查询结果，使用流都可以提供高效的数据处理方式。

🚀Node.js流的重点知识

以下是关于Node.js流的重点知识点总结：

1. 流的类型：流可以是可读的、可写的，或者可读可写的。在Node.js中，有四种内置的流类型，
   1. 可读流（Readable）、
   2. 可写流（Writable）、
   3. 双工流（Duplex）
   4. 转换流（Transform）。
2. 常见的流事件：所有的流都是EventEmitter类的实例，可以通过事件和方法来处理流的数据。
   1. data（当有数据可读时触发）、
   2. end（当没有更多数据可读时触发）
   3. error（当发生错误时触发）等。
3. 常见的流方法包括
   1. write（向流中写入数据）
   2. read（从流中读取数据）
   3. pipe（将一个可读流的数据传输到一个可写流）等。
4. 流的工作原理：流将要传输的数据处理成连续的块（chunk），通过一小块一小块地传输数据，从而降低内存消耗并提高性能。这种逐段处理的方式使得流非常适合处理大量数据或需要逐段处理的场景。
5. 内置的流对象：Node.js中有许多内置的流对象，如HTTP请求、文件读写、标准输入输出（stdin/stdout）等。这些内置的流对象可以直接使用，无需直接实现流的接口。
6. stream模块：stream模块是用于构建实现流接口的对象的模块，可以通过require('stream')导入。开发人员可以使用stream模块创建自定义的流实例，实现更复杂的流操作。

🌈 可读流

🚀什么是可读流

可读流是对提供数据的来源的一种抽象。可读流的例子包括客户端的 HTTP 响应、服务器的 HTTP请求、fs 的读取流、zlib 流、crypto 流、TCP Socket、子进程 stdout与stderr、process.stdin 等。所有可读流都实现了 stream.Readable 类定义的接口。

fs读取文件案例

文件：fs读取文件.js

const fs = require("fs");
// 创建可读流
const rs = fs.createReadStream("./movie.txt");

let content = "";
// 当有数据可读时触发"data"事件
rs.on("data", chunk => {
    // 将读取的数据块拼接到content变量中
    content += chunk;
});

// 当可读流结束时触发"end"事件
rs.on("end", () => {
    // 所有数据已读取完毕，输出content内容
    console.log(content);
});
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

文件：movie.txt

The best time to plant a tree is 20 years ago. 
The second-best time is now.
种一棵树最好的时间是20年前，
其次是现在
1
2
3
4

输出结果

PS D:\WuWorkSpace\code\NodejsProject\nodejs实战学习\【第2章  Node.js基础】\2.7 Node的流> node .\fs可读流.js
The best time to plant a tree is 20 years ago.
The second-best time is now.
种一棵树最好的时间是20年前，
其次是现在
1
2
3
4
5

process.stdin案例

 process.stdin.pipe(process.stdout);
1

这段代码使用了Node.js的process.stdin和process.stdout流对象，并通过pipe方法将输入流(process.stdin)的数据直接传输到输出流(process.stdout)中。

具体解析如下：

• process.stdin是一个可读流，它表示标准输入流（stdin），可以用于接收用户的输入。
• process.stdout是一个可写流，它表示标准输出流（stdout），可以用于向控制台输出数据。

通过调用pipe方法，我们将输入流(process.stdin)连接到输出流(process.stdout)，这样输入流中的数据会被自动传输到输出流中。

这段代码的作用是将用户在控制台输入的内容直接输出到控制台，实现了一个简单的输入输出的管道（复读机）。

动图演示如下

http案例

客户端的 HTTP 响应、服务器的 HTTP请求是可读流。

服务端文件：http-server.js

const http = require("http");
const server = http.createServer((req,res)=>{
    if(req.method === "POST" && req.url.includes("/upload")){
        process.stdout.write("服务端收到了请求：");

        // req 服务器的 HTTP 请求（可读流）
        req.pipe(process.stdout);
        // res 服务器的 HTTP 响应（可写流）
        res.write("我是服务端响应");
        res.end();
    }else{
        res.writeHead(404);
        res.end("Not Found");
    }
})
server.listen(9800);
console.log(`服务已经运行在：http://localhost:9800`);
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

客户端代码文件：http-client.js

const http = require("http");
const options = {
    hostname : "127.0.0.1",
    port : 9800,
    path : "/upload",
    method : "POST"
};
const req = http.request(options,res=>{
    process.stdout.write("客户端获取响应：");

    // res 客户端的 HTTP 响应（可读流）
    res.pipe(process.stdout);
})

// req 客户端的 HTTP 请求（可写流）
req.write("我是客户端。");
req.end();
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

以下是运行演示：先运行服务端代码 http-server.js ，再运行客户端代码http-client.js。

$ node .\http-client.js
客户端获取响应：我是服务端响应
1
2

$ node .\http-server.js
服务已经运行在：http://localhost:9800
服务端收到了请求：我是客户端。
1
2
3

··

🚀可读流de两种模式

可读流有两种模式:流动(Flowing)和暂停(Paused )。在流动模式中，数据被自动从底层系统读取，并通过 EventEmitter 接口的事件尽可能快地提供给应用程序。在暂停模式中，必须显式调用stream.read()方法读取数据块。
所有可读流一开始都处于暂停模式，,之后可以通过以下方式切换到流动模式。

• 添加data 事件处理函数。
• 调用stream.resume()方法。
• 调用 stream.pipe()方法。

可读流也可以通过以下方式切换回暂停模式。

• 如果没有管道目标，则调用 stream.pause()方法
• 如果有管道目标，则移除所有管道目标。调用 stream.unpipe()方法可以移除多个管道目标。

注意​ 👉: ​ 只有提供了数据消费或忽略数据的机制后，可读流才会产生数据。如果消费的机制被禁用或移除，则可读流会停止产生数据。

stream.Readable类定义的主要事件:

• data：当有数据可读时被触发。
• end：没有更多的数据可读时被触发。
• close：当流或其底层资源被关闭时被触发。

stream.Readable 类定义的主要方法

• readable.read([size])：从内部缓冲区拉取并返回数据。
• readable.pause()：使流动模式的流停止触发data事件，并切换出流动模式。
• readable.setEncoding(encoding)：为从可读流读取的数据设置字符编码。