Node篇

Posted 2022-01-04 still1

【Node篇】

Node.js中的stream（流）- 基础篇

1）什么是stream（流）

流（stream）在 Node.js 中是处理流数据的抽象接口（abstract interface）。 stream 模块提供了基础的 API 。使用这些 API 可以很容易地来构建实现流接口的对象。

流是可读的、可写的，或是可读写的。

2）NodeJs中的Stream的几种类型

Node.js 中有四种基本的流类型：

Readable - 可读的流（fs.createReadStream()）

Writable - 可写的流（fs.createWriteStream()）

Duplex - 可读写的流（net.Socket）

Transform - 在读写过程中可以修改和变换数据的 Duplex 流 (例如 zlib.createDeflate())

NodeJs中关于流的操作被封装到了Stream模块中，这个模块也被多个核心模块所引用。

const stream = require(‘stream‘);

在 NodeJS 中对文件的处理多数使用流来完成。

普通文件

设备文件（stdin、stdout）

网络文件（http、net）

 

注：在NodeJs中所有的Stream(流)都是EventEmitter的实例

 

Example：

 

1.将1.txt的文件内容读取为流数据

const fs = require(‘fs‘);

// 创建一个可读流（生产者）
let rs = fs.createReadStream(‘./1.txt‘); 

通过fs模块提供的createReadStream()可以轻松创建一个可读的文件流。但我们并有直接使用Stream模块，因为fs模块内部已经引用了Stream模块并做了封装。所以说 流（stream）在 Node.js 中是处理流数据的抽象接口，提供了基础Api来构建实现流接口的对象。

var rs = fs.createReadStream(path,[options]);

1.path 读取文件的路径

2.options

• flags打开文件的操作, 默认为’r’
• mode 权限位 0o666
• encoding默认为null
• start开始读取的索引位置
• end结束读取的索引位置(包括结束位置)
• highWaterMark读取缓存区默认的大小64kb

Node.js 提供了多种流对象。 例如：

• HTTP 请求 （request response）
• process.stdout 就都是流的实例。

2.创建可写流（消费者）处理可读流

将1.txt的可读流 写入到2.txt文件中 这时我们需要一个可写流

const fs = require(‘fs‘);
// 创建一个可写流
let ws = fs.createWriteStream(‘./2.txt‘);
// 通过pipe让可读流流入到可写流 写入文件
rs.pipe(ws); 

var ws = fs.createWriteStream(path,[options]);

1.path 读取文件的路径

2.options

• flags打开文件的操作, 默认为’w’
• mode 权限位 0o666
• encoding默认为utf8
• autoClose:true是否自动关闭文件
• highWaterMark读取缓存区默认的大小16kb

pipe 它是Readable流的方法，相当于一个”管道”，数据必须从上游 pipe 到下游，也就是从一个 readable 流 pipe 到 writable 流。 

后续将深入将介绍pipe。

如上图，我们把文件比作装水的桶，而水就是文件里的内容，我们用一根管子(pipe)连接两个桶使得水从一个桶流入另一个桶，这样就慢慢的实现了大文件的传输过程。

3）为什么应该使用Stream

当有用户在线看视频，假定我们通过HTTP请求返回给用户视频内容

const http = require(‘http‘);
const fs = require(‘fs‘);

http.createServer((req, res) => 
    fs.readFile(videoPath, (err, data) => 
        res.end(data);
    );
).listen(8080);

但这样有两个明显的问题

1.视频文件需要全部读取完，才能返回给用户，这样等待时间会很长 

2.视频文件一次全放入内存中，内存吃不消

用流可以将视频文件一点一点读到内存中，再一点一点返回给用户，读一部分，写一部分。（利用了 HTTP 协议的 Transfer-Encoding: chunked 分段传输特性），用户体验得到优化，同时对内存的开销明显下降

const http = require(‘http‘);
const fs = require(‘fs‘);

http.createServer((req, res) => 
    fs.createReadStream(videoPath).pipe(res);
).listen(8080);

4）可读流（Readable Stream）

可读流（Readable streams）是对提供数据的源头（source）的抽象。

例如:

HTTP responses, on the client

HTTP requests, on the server

fs read streams

TCP sockets

process.stdin

 

所有的 Readable 都实现了 stream.Readable 类定义的接口。

 

可读流的两种模式（flowing 和 paused）

1.在 flowing 模式下， 可读流自动从系统底层读取数据，并通过 EventEmitter 接口的事件尽快将数据提供给应用。

 

2.在 paused 模式下，必须显式调用 stream.read()方法来从流中读取数据片段。

 

所有初始工作模式为paused的Readable流，可以通过下面三种途径切换为flowing模式：

监听’data’事件

调用stream.resume()方法

调用stream.pipe()方法将数据发送到Writable

 

流动模式flowing

流切换到流动模式 监听data事件

const rs = fs.createReadStream(‘./1.txt‘);
const ws = fs.createWriteStream(‘./2.txt‘);
rs.on(‘data‘, chunk => 
    ws.write(chunk);
);
ws.on(‘end‘, () => 
    ws.end();
);

如果写入的速度跟不上读取的速度，有可能导致数据丢失。正常的情况应该是，写完一段，再读取下一段，如果没有写完的话，就让读取流先暂停，等写完再继续。

var fs = require(‘fs‘);
// 读取highWaterMark(3字节)数据，读完之后填充缓存区，然后触发data事件
var rs = fs.createReadStream(sourcePath, 
    highWaterMark: 3 
);
var ws = fs.createWriteStream(destPath, 
    highWaterMark: 3
);

rs.on(‘data‘, function(chunk)  // 当有数据流出时，写入数据
    if (ws.write(chunk) === false)  // 如果没有写完，暂停读取流
        rs.pause();
    
);

ws.on(‘drain‘, function()  // 缓冲区清空触发drain事件 这时再继续读取
    rs.resume();
);

rs.on(‘end‘, function()  // 当没有数据时，关闭数据流
    ws.end();
);

或者使用更直接的pipe
fs.createReadStream(sourcePath).pipe(fs.createWriteStream(destPath));

暂停模式paused

1.在流没有 pipe() 时，调用 pause() 方法可以将流暂停 

2.pipe() 时，需要移除所有 data 事件的监听，再调用 unpipe() 方法

 

read(size)

流在暂停模式下需要程序显式调用 read() 方法才能得到数据。read() 方法会从内部缓冲区中拉取并返回若干数据，当没有更多可用数据时，会返回null。read()不会触发’data’事件。

 

使用 read() 方法读取数据时，如果传入了 size 参数，那么它会返回指定字节的数据；当指定的size字节不可用时，则返回null。如果没有指定size参数，那么会返回内部缓冲区中的所有数据。 

NodeJS 为我们提供了一个 readable 的事件，事件在可读流准备好数据的时候触发，也就是先监听这个事件，收到通知又数据了我们再去读取就好了：

const fs = require(‘fs‘);
rs = fs.createReadStream(sourcePath);

// 当你监听 readable事件的时候，会进入暂停模式
rs.on(‘readable‘, () => 
    console.log(rs._readableState.length);
        // read如果不加参数表示读取整个缓存区数据
        // 读取一个字段,如果可读流发现你要读的字节小于等于缓存字节大小，则直接返回
        let ch = rs.read(1);
);

暂停模式 缓存区的数据以链表的形式保存在BufferList中

 

5) 可写流（Writable Stream）

可写流是对数据流向设备的抽象，用来消费上游流过来的数据，通过可写流程序可以把数据写入设备，常见的是本地磁盘文件或者 TCP、HTTP 等网络响应。

 

Writable 的例子包括了：

HTTP requests, on the client

HTTP responses, on the server

fs write streams

zlib streams

crypto streams

TCP sockets

child process stdin

process.stdout, process.stderr

所有 Writable 流都实现了 stream.Writable 类定义的接口。

 

process.stdin.pipe(process.stdout);

process.stdout 是一个可写流，程序把可读流 process.stdin 传过来的数据写入的标准输出设备。在了解了可读流的基础上理解可写流非常简单，流就是有方向的数据，其中可读流是数据源，可写流是目的地，中间的管道环节是双向流。

 

可写流使用
调用可写流实例的 write() 方法就可以把数据写入可写流

const fs = require(‘fs‘);
const rs = fs.createReadStream(sourcePath);
const ws = fs.createWriteStream(destPath);

rs.setEncoding(‘utf-8‘); // 设置编码格式
rs.on(‘data‘, chunk => 
  ws.write(chunk); // 写入数据
);

监听了可读流的 data 事件就会使可读流进入流动模式，我们在回调事件里调用了可写流的 write() 方法，这样数据就被写入了可写流抽象的设备destPath中。

write() 方法有三个参数：

chunk String| Buffer，表示要写入的数据

encoding 当写入的数据是字符串的时候可以设置编码

callback 数据被写入之后的回调函数

‘drain’事件

如果调用 stream.write(chunk) 方法返回 false，表示当前缓存区已满，流将在适当的时机（缓存区清空后）触发 ‘drain

const fs = require(‘fs‘);
const rs = fs.createReadStream(sourcePath);
const ws = fs.createWriteStream(destPath);

rs.setEncoding(‘utf-8‘); // 设置编码格式
rs.on(‘data‘, chunk => 
  let flag = ws.write(chunk); // 写入数据
  if (!flag)  // 如果缓存区已满暂停读取
      rs.pause();
  
);

ws.on(‘drain‘, () => 
    rs.resume(); // 缓存区已清空 继续读取写入
);

 

6总结

stream(流)分为可读流（flowing mode 和 paused mode）、可写流、可读写流，Node.js 提供了多种流对象。 例如， HTTP 请求 和 process.stdout 就都是流的实例。stream 模块提供了基础的 API 。使用这些 API 可以很容易地来构建实现流接口的对象。它们底层都调用了stream模块并进行封装。