Node.js中StreamAPI的使用

基本介绍

在 Node.js 中&＃xff0c;读取文件的方式有两种&＃xff0c;一种是用 fs.readFile&＃xff0c;另外一种是利用 fs.createReadStream 来读取。

fs.readFile 对于每个 Node.js 使用者来说最熟悉不过了&＃xff0c;简单易懂&＃xff0c;很好上手。但它的缺点是会先将数据全部读入内存&＃xff0c;一旦遇到大文件的时候&＃xff0c;这种方式读取的效率就非常低下了。

而 fs.createReadStream 则是通过 Stream 来读取数据&＃xff0c;它会把文件&＃xff08;数据&＃xff09;分割成小块&＃xff0c;然后触发一些特定的事件&＃xff0c;我们可以监听这些事件&＃xff0c;编写特定的处理函数。这种方式相对上面来说&＃xff0c;并不好上手&＃xff0c;但它效率非常高。

事实上&＃xff0c; Stream 在 Node.js 中并非仅仅用在文件处理上&＃xff0c;其他地方也可以看到它的身影&＃xff0c;如 process.stdin/stdout, http, tcp sockets, zlib, crypto 等都有用到。

本文是我学习 Node.js 中的 Stream API 中的一点总结&＃xff0c;希望对大家有用。

特点

基于事件通讯
可以通过 pipe 来连接流

种类

Readable Stream 可读数据流
Writeable Stream 可写数据流
Duplex Stream 双向数据流&＃xff0c;可以同时读和写
Transform Stream 转换数据流&＃xff0c;可读可写&＃xff0c;同时可以转换&＃xff08;处理&＃xff09;数据

事件

可读数据流的事件

readable 数据向外流时触发
data 对于那些没有显式暂停的数据流&＃xff0c;添加data事件监听函数&＃xff0c;会将数据流切换到流动态&＃xff0c;尽快向外提供数据
end 读取完数据时触发。注意不能和 writeableStream.end() 混淆&＃xff0c;writeableStream 并没有 end 事件&＃xff0c;只有 .end() 方法
close 数据源关闭时触发
error 读取数据发生错误时触发

可写数据流的事件

drain writable.write(chunk) 返回 false 之后&＃xff0c;缓存全部写入完成&＃xff0c;可以重新写入时就会触发
finish 调用 .end 方法时&＃xff0c;所有缓存的数据释放后触发&＃xff0c;类似于可读数据流中的 end 事件&＃xff0c;表示写入过程结束
pipe 作为 pipe 目标时触发
unpipe 作为 unpipe 目标时触发
error 写入数据发生错误时触发

状态

可读数据流有两种状态&＃xff1a;流动态 和 暂停态&＃xff0c;改变数据流状态的方法如下&＃xff1a;

暂停态 -> 流动态

添加 data 事件的监听函数
调用 resume 方法
调用 pipe 方法

注意&＃xff1a; 如果转为流动态时&＃xff0c;没有 data 事件的监听函数&＃xff0c;也没有 pipe 方法的目的地&＃xff0c;那么数据将遗失。

流动态 -> 暂停态

不存在 pipe 方法的目的地时&＃xff0c;调用 pause 方法
存在 pipe 方法的目的地时&＃xff0c;移除所有 data 事件的监听函数&＃xff0c;并且调用 unpipe 方法&＃xff0c;移除所有 pipe 方法的目的地

注意&＃xff1a; 只移除 data 事件的监听函数&＃xff0c;并不会自动引发数据流进入「暂停态」。另外&＃xff0c;存在 pipe 方法的目的地时&＃xff0c;调用 pause 方法&＃xff0c;并不能保证数据流总是处于暂停态&＃xff0c;一旦那些目的地发出数据请求&＃xff0c;数据流有可能会继续提供数据。

用法

读写文件

var fs &＃61; require(&＃39;fs&＃39;); // 新建可读数据流 var rs &＃61; fs.createReadStream(&＃39;./test1.txt&＃39;); // 新建可写数据流 var ws &＃61; fs.createWriteStream(&＃39;./test2.txt&＃39;);// 监听可读数据流结束事件 rs.on(&＃39;end&＃39;, function() {console.log(&＃39;read text1.txt successfully!&＃39;); }); // 监听可写数据流结束事件 ws.on(&＃39;finish&＃39;, function() {console.log(&＃39;write text2.txt successfully!&＃39;); }); // 把可读数据流转换成流动态&＃xff0c;流进可写数据流中 rs.pipe(ws);

读取 CSV 文件&＃xff0c;并上传数据&＃xff08;我在生产环境中写过&＃xff09;

var fs &＃61; require(&＃39;fs&＃39;); var es &＃61; require(&＃39;event-stream&＃39;); var csv &＃61; require(&＃39;csv&＃39;); var parser &＃61; csv.parse(); var transformer &＃61; csv.transform(function(record) {return record.join(&＃39;,&＃39;); });var data &＃61; fs.createReadStream(&＃39;./demo.csv&＃39;); data.pipe(parser).pipe(transformer)// 处理前一个 stream 传递过来的数据.pipe(es.map(function(data, callback) {upload(data, function(err) {callback(err);});}))// 相当于监听前一个 stream 的 end 事件.pipe(es.wait(function(err, body) {process.stdout.write(&＃39;done!&＃39;);}));

常见坑

用 rs.pipe(ws) 的方式来写文件并不是把 rs 的内容 append 到 ws 后面&＃xff0c;而是直接用 rs 的内容覆盖 ws 原有的内容
已结束/关闭的流不能重复使用&＃xff0c;必须重新创建数据流
pipe 方法返回的是目标数据流&＃xff0c;如 a.pipe(b) 返回的是 b&＃xff0c;因此监听事件的时候请注意你监听的对象是否正确
- 如果你要监听多个数据流&＃xff0c;同时你又使用了 pipe 方法来串联数据流的话&＃xff0c;你就要写成&＃xff1a;
  
  data.on(&＃39;end&＃39;, function() {console.log(&＃39;data end&＃39;);}).pipe(a).on(&＃39;end&＃39;, function() {console.log(&＃39;a end&＃39;);}).pipe(b).on(&＃39;end&＃39;, function() {console.log(&＃39;b end&＃39;);});

常用类库

event-stream 用起来有函数式编程的感觉&＃xff0c;个人比较喜欢
awesome-nodejs#streams 由于其他 stream 库我都没用过&＃xff0c;所以有需求的就直接看这里吧

出处

https://scarletsky.github.io/2015/10/22/node-stream-api-learning/

参考资料

阮一峰 - stream接口
nodejs.org Stream
Transforming data with Node.js transform streams
NodeJS: What&＃39;s the difference between a Duplex stream and a Transform stream?