node.js stream（流）的使用

node.js stream（流）的使用

• 什么是 stream ？
• stream 的作用
• 可读流实例
• 可写流实例
• 管道流（pipe）实例
• `新手学习中，如果发现哪一部分讲解的有问题，欢迎评论区留言指正。`

什么是 stream ？

官方文档：stream 是 Node.js 中处理流式数据的抽象接口，Node 中有很多对象实现了这个接口。例如，对http 服务器发起请求的 request 对象等。

我的理解是stream就像是一条带有闸门的水流，在进行IO操作时，操作的数据就是水流，内存就是管道，我们可以通过控制闸门升降来控制水流的大小，从而控制管道内的水量。

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

• Writable - 可写入数据的流（例如 fs.createWriteStream()）。
• Readable - 可读取数据的流（例如 fs.createReadStream()）。
• Duplex - 可读又可写的流（例如 net.Socket）。
• Transform - 在读写过程中可以修改或转换数据的 Duplex 流（例如 zlib.createDeflate()）。

所有的 Stream 对象都是 EventEmitter 的实例（这里有解释）。常用的事件有：

• data - 当有数据可读时触发。

• end - 没有更多的数据可读时触发。

• error - 在接收和写入过程中发生错误时触发。

• finish - 所有数据已被写入到底层系统时触发。

参考：node.js中文文档

stream 的作用

stream API 的主要目标，特别是 stream.pipe()，是为了限制数据的缓冲到可接受的程度，也就是读写速度不一致的源头与目的地不会压垮内存。（另外 stream 最大的作用是读取大文件。）

例如：

• 使用 readFile 读取文件的时候，会一次性把文件读到内存中去，如果文件过大，会对系统造成一定压力，甚至崩溃。

• 使用 stream 读取文件，不会一次性的读入到内存中。每次只会读取数据源的一个数据块，然后后续过程中可以立即处理该数据块（数据处理完成后会进入垃圾回收机制），而不用等待所有的数据。

可读流实例

可读流是对提供数据的来源的一种抽象。

可读流的例子包括：

• 客户端的 HTTP 响应
• 服务器的 HTTP 请求
• fs 的读取流
• zlib 流
• crypto 流
• TCP socket
• 子进程 stdout 与 stderr
• process.stdin

所有可读流都实现了 stream.Readable 类定义的接口（可参考官方文档上stream.Readable有哪些方法、属性，以及事件）。

这里以 fs 模块的 createReadStream 函数为例（createReadStream 官方文档）：

const fs = require('fs');

// demo.txt中内容：this is a demo
const file = fs.createReadStream('./demo.txt', {
flags: 'r', // 文件的操作方式，默认是 r
encoding: 'utf-8', // 编码格式
start: 0, // 开始读取的位置
end: 6, // 结束读取的位置
highWaterMark: 1 // 每次读取的个数, 以字节为单位, 默认值64 * 1024字节
});

// 每当从流中读取到数据时就会触发 data 事件
file.on('data', chuck => {
console.log(`读到数据---${chuck}`);
});

// close事件在流关闭后触发，这里故意和 end 事件调换了定义的顺序，观察结果会不会发生变化
file.on('close', () => {
console.log('读取流已经关闭');
});

// end 事件在数据读取完成时才会触发
file.on('end', () => {
console.log('已经读取完成');
});

// 必须为 error 事件添加监听器，否则程序在遇到异常时会终止运行
file.on('error', err => {
console.error(err);
});

out:
读到数据---t
读到数据---h
读到数据---i
读到数据---s
读到数据---
读到数据---i
读到数据---s
已经读取完成
读取流已经关闭

可写流实例

可写流是对数据要被写入的目的地的一种抽象。

可写流的例子包括：

• 客户端的 HTTP 响应
• 服务器的 HTTP 请求
• fs 的写入流
• zlib 流
• crypto 流
• TCP socket
• 子进程 stdin
• process.stdout、process.stderr

所有可写流都实现了 stream.Writable 类定义的接口（可参考官方文档上stream.Writable有哪些方法、属性，以及事件）。

这里以 fs 模块的 createWriteStream函数为例（createWriteStream官方文档）：

const fs = require('fs');

const file = fs.createWriteStream('./demo01.txt', {
flags: 'w',  // 操作方式，默认是 w
encoding: 'utf-8',  // 编码格式
start: 0,  // 在文件中开始写入的位置
highWaterMark: 3  // 每次写入的字节数, 一般不需要配置
});

let f = file.write('a', 'utf-8', () => {
console.log("写入数据a");
});

console.log(f);

f = file.write('bb', 'utf-8', () => {
console.log("写入数据bb");
});

console.log(f);

// 标记文件末尾
file.end();

// end 方法执行完成才会触发 finish 事件
file.on('finish', () => {
console.log("全部数据写入完成");
})

// 必须为 error 事件添加监听器，否则程序在遇到异常时会终止运行
file.on('error', err => {
console.error(err);
})

out:
true
false
写入数据a
写入数据bb
全部数据写入完成

writable.write() 写入数据到流，流在接收了数据后，如果内部的缓冲小于创建流时配置的 highWaterMark，则返回 true。如果返回 false，则应该停止向流写入数据，直到 ‘drain’ 事件被触发。

解决上诉问题的方法是封装一个write函数：

function write(data, cb) {
if (!stream.write(data)) {
stream.once('drain', cb);
} else {
process.nextTick(cb);
}
}

// 在回调函数被执行后再进行其他的写入。
write('hello', () => {
console.log('完成写入，可以进行更多的写入');
});

管道流（pipe）实例

管道流（pipe）官方文档

readable.pipe() 方法绑定可写流到可读流，将可读流自动切换到流动模式，并将可读流的所有数据推送到绑定的可写流。 数据流会被自动管理，所以即使可读流更快，目标可写流也不会超负荷。

管道提供了一个输出流到输入流的机制，通常我们用于从一个流中获取数据并将数据传递到另外一个流中。

例如：我们需要把我们上面可读流读到的数据需要放到可写流中写入到文件里

const fs = require('fs');

const fRead = fs.createReadStream('./demo.txt', {
flags: 'r',
encoding: 'utf-8',
highWaterMark: 1
});

// 第二个参数可以直接传入一个表示编码的字符串
const fWrite = fs.createWriteStream('demo01.txt', "utf-8");

// 利用管道连接可读流与可写流
fRead.pipe(fWrite);

// 这里只是为了验证管道的写入方式
fRead.on('data', (chunk) => {
console.log(`写入${chunk}`);
})

// 为可读流的 error 事件注册监听器
fRead.on('error', err => {
console.error(err);

// 手动关闭可写流
fWrite.end();
})
// 为可写流的 error 事件注册监听器
fWrite.on('error',err => {
console.error(err);
})

out:
写入d
写入e
写入m
写入o
写入!

从结果可以看出来，可写流每次写入文件的大小等于可读流设置的 highWaterMark。

上面的例子中 highWaterMark 设置为1，即每次读取一个字节，写入一个字节。

注意：

1. 默认情况下，当来源可读流触发 ‘end’ 事件时，目标可写流也会调用 stream.end() 结束写入。若要禁用这种默认行为， end 选项应设为 false，这样目标流就会保持打开。

   fRead.pipe(fWrite, { end: false });
   fRead.on('end', () => {
   fWrite.end('结束');
   });

2. 如果可读流在处理期间发送错误，则可写流目标不会自动关闭。 如果发生错误，则需要手动关闭每个流以防止内存泄漏。

   fRead.on('error', err => {
   console.error(err);
   
   // 手动关闭可写流
   fWrite.end();
   })

新手学习中，如果发现哪一部分讲解的有问题，欢迎评论区留言指正。