彻底掌握 Node.js 四大流,解决爆缓冲区的“背压”问题

时间:2022-10-24 09:39:07

彻底掌握 Node.js 四大流,解决爆缓冲区的“背压”问题

把一个东西从 A 搬到 B 该怎么搬呢?

抬起来,移动到目的地,放下不就行了么。

那如果这个东西有一吨重呢?

那就一部分一部分的搬。

其实 IO 也就是搬东西,包括网络的 IO、文件的 IO,如果数据量少,那么直接传送全部内容就行了,但如果内容特别多,一次性加载到内存会崩溃,而且速度也慢,这时候就可以一部分一部分的处理,这就是流的思想。

各种语言基本都实现了 stream 的 api,Node.js 也是,stream api 是比较常用的,下面我们就来探究一下 stream。

本文会回答以下问题:

  • Node.js 的 4 种 stream 是什么
  • 生成器如何与 Readable Stream 结合
  • stream 的暂停和流动
  • 什么是背压问题,如何解决

Node.js 的 4种 stream

流的直观感受

从一个地方流到另一个地方,显然有流出的一方和流入的一方,流出的一方就是可读流(readable),而流入的一方就是可写流(writable)。

彻底掌握 Node.js 四大流,解决爆缓冲区的“背压”问题

当然,也有的流既可以流入又可以流出,这种叫做双工流(duplex)

彻底掌握 Node.js 四大流,解决爆缓冲区的“背压”问题

既然可以流入又可以流出,那么是不是可以对流入的内容做下转换再流出呢,这种流叫做转换流(transform)

彻底掌握 Node.js 四大流,解决爆缓冲区的“背压”问题

duplex 流的流入和流出内容不需要相关,而 transform 流的流入和流出是相关的,这是两者的区别。

流的 api

Node.js 提供的 stream 就是上面介绍的那 4 种:

  1. conststream=require('stream');
  2.  
  3. //可读流
  4. constReadable=stream.Readable;
  5. //可写流
  6. constWritable=stream.Writable;
  7. //双工流
  8. constDuplex=stream.Duplex;
  9. //转换流
  10. constTransform=stream.Transform;

它们都有要实现的方法:

  • Readable 需要实现 _read 方法来返回内容
  • Writable 需要实现 _write 方法来接受内容
  • Duplex 需要实现 _read 和 _write 方法来接受和返回内容
  • Transform 需要实现 _transform 方法来把接受的内容转换之后返回

我们分别来看一下:

Readable

Readable 要实现 _read 方法,通过 push 返回具体的数据。

  1. constStream=require('stream');
  2.  
  3. constreadableStream=Stream.Readable();
  4.  
  5. readableStream._read=function(){
  6. this.push('阿门阿前一棵葡萄树,');
  7. this.push('阿东阿东绿的刚发芽,');
  8. this.push('阿东背着那重重的的壳呀,');
  9. this.push('一步一步地往上爬。')
  10. this.push(null);
  11. }
  12.  
  13. readableStream.on('data',(data)=@gt;{
  14. console.log(data.toString())
  15. });
  16.  
  17. readableStream.on('end',()=@gt;{
  18. console.log('done~');
  19. });

当 push 一个 null 时,就代表结束流。

执行效果如下:

彻底掌握 Node.js 四大流,解决爆缓冲区的“背压”问题

创建 Readable 也可以通过继承的方式:

  1. constStream=require('stream');
  2.  
  3. classReadableDongextendsStream.Readable{
  4.  
  5. constructor(){
  6. super();
  7. }
  8.  
  9. _read(){
  10. this.push('阿门阿前一棵葡萄树,');
  11. this.push('阿东阿东绿的刚发芽,');
  12. this.push('阿东背着那重重的的壳呀,');
  13. this.push('一步一步地往上爬。')
  14. this.push(null);
  15. }
  16.  
  17. }
  18.  
  19. constreadableStream=newReadableDong();
  20.  
  21. readableStream.on('data',(data)=@gt;{
  22. console.log(data.toString())
  23. });
  24.  
  25. readableStream.on('end',()=@gt;{
  26. console.log('done~');
  27. });

可读流是生成内容的,那么很自然可以和生成器结合:

  1. constStream=require('stream');
  2.  
  3. classReadableDongextendsStream.Readable{
  4.  
  5. constructor(iterator){
  6. super();
  7. this.iterator=iterator;
  8. }
  9.  
  10. _read(){
  11. constnext=this.iterator.next();
  12. if(next.done){
  13. returnthis.push(null);
  14. }else{
  15. this.push(next.value)
  16. }
  17. }
  18.  
  19. }
  20.  
  21. function*songGenerator(){
  22. yield'阿门阿前一棵葡萄树,';
  23. yield'阿东阿东绿的刚发芽,';
  24. yield'阿东背着那重重的的壳呀,';
  25. yield'一步一步地往上爬。';
  26. }
  27.  
  28. constsongIterator=songGenerator();
  29.  
  30. constreadableStream=newReadableDong(songIterator);
  31.  
  32. readableStream.on('data',(data)=@gt;{
  33. console.log(data.toString())
  34. });
  35.  
  36. readableStream.on('end',()=@gt;{
  37. console.log('done~');
  38. });

这就是可读流,通过实现 _read 方法来返回内容。

Writable

Writable 要实现 _write 方法,接收写入的内容。

  1. constStream=require('stream');
  2.  
  3. constwritableStream=Stream.Writable();
  4.  
  5. writableStream._write=function(data,enc,next){
  6. console.log(data.toString());
  7. //每秒写一次
  8. setTimeout(()=@gt;{
  9. next();
  10. },1000);
  11. }
  12.  
  13. writableStream.on('finish',()=@gt;console.log('done~'));
  14.  
  15. writableStream.write('阿门阿前一棵葡萄树,');
  16. writableStream.write('阿东阿东绿的刚发芽,');
  17. writableStream.write('阿东背着那重重的的壳呀,');
  18. writableStream.write('一步一步地往上爬。');
  19. writableStream.end();

接收写入的内容,打印出来,并且调用 next 来处理下一个写入的内容,这里调用 next 是异步的,可以控制频率。

跑了一下,确实可以正常的处理写入的内容:

彻底掌握 Node.js 四大流,解决爆缓冲区的“背压”问题

这就是可写流,通过实现 _write 方法来处理写入的内容。

Duplex

Duplex 是可读可写,同时实现 _read 和 _write 就可以了

  1. constStream=require('stream');
  2.  
  3. varduplexStream=Stream.Duplex();
  4.  
  5. duplexStream._read=function(){
  6. this.push('阿门阿前一棵葡萄树,');
  7. this.push('阿东阿东绿的刚发芽,');
  8. this.push('阿东背着那重重的的壳呀,');
  9. this.push('一步一步地往上爬。')
  10. this.push(null);
  11. }
  12.  
  13. duplexStream._write=function(data,enc,next){
  14. console.log(data.toString());
  15. next();
  16. }
  17.  
  18. duplexStream.on('data',data=@gt;console.log(data.toString()));
  19. duplexStream.on('end',data=@gt;console.log('readdone~'));
  20.  
  21. duplexStream.write('阿门阿前一棵葡萄树,');
  22. duplexStream.write('阿东阿东绿的刚发芽,');
  23. duplexStream.write('阿东背着那重重的的壳呀,');
  24. duplexStream.write('一步一步地往上爬。');
  25. duplexStream.end();
  26.  
  27. duplexStream.on('finish',data=@gt;console.log('writedone~'));

整合了 Readable 流和 Writable 流的功能,这就是双工流 Duplex。

彻底掌握 Node.js 四大流,解决爆缓冲区的“背压”问题

Transform

Duplex 流虽然可读可写,但是两者之间没啥关联,而有的时候需要对流入的内容做转换之后流出,这时候就需要转换流 Transform。

Transform 流要实现 _transform 的 api,我们实现下对内容做反转的转换流:

  1. constStream=require('stream');
  2.  
  3. classTransformReverseextendsStream.Transform{
  4.  
  5. constructor(){
  6. super()
  7. }
  8.  
  9. _transform(buf,enc,next){
  10. constres=buf.toString().split('').reverse().join('');
  11. this.push(res)
  12. next()
  13. }
  14. }
  15.  
  16. vartransformStream=newTransformReverse();
  17.  
  18. transformStream.on('data',data=@gt;console.log(data.toString()))
  19. transformStream.on('end',data=@gt;console.log('readdone~'));
  20.  
  21. transformStream.write('阿门阿前一棵葡萄树');
  22. transformStream.write('阿东阿东绿的刚发芽');
  23. transformStream.write('阿东背着那重重的的壳呀');
  24. transformStream.write('一步一步地往上爬');
  25. transformStream.end()
  26.  
  27. transformStream.on('finish',data=@gt;console.log('writedone~'));

跑了一下,效果如下:

彻底掌握 Node.js 四大流,解决爆缓冲区的“背压”问题

流的暂停和流动

我们从 Readable 流中获取内容,然后流入 Writable 流,两边分别做 _read 和 _write 的实现,就实现了流动。

彻底掌握 Node.js 四大流,解决爆缓冲区的“背压”问题

背压

但是 read 和 write 都是异步的,如果两者速率不一致呢?

如果 Readable 读入数据的速率大于 Writable 写入速度的速率,这样就会积累一些数据在缓冲区,如果缓冲的数据过多,就会爆掉,会丢失数据。

而如果 Readable 读入数据的速率小于 Writable 写入速度的速率呢?那没关系,最多就是中间有段空闲时期。

这种读入速率大于写入速率的现象叫做“背压”,或者“负压”。也很好理解,写入段压力比较大,写不进去了,会爆缓冲区,导致数据丢失。

这个缓冲区大小可以通过 readableHighWaterMark 和 writableHightWaterMark 来查看,是 16k。

彻底掌握 Node.js 四大流,解决爆缓冲区的“背压”问题

解决背压

怎么解决这种读写速率不一致的问题呢?

当没写完的时候,暂停读就行了。这样就不会读入的数据越来越多,驻留在缓冲区。

readable stream 有个 readableFlowing 的属性,代表是否自动读入数据,默认为 true,也就是自动读入数据,然后监听 data 事件就可以拿到了。

当 readableFlowing 设置为 false 就不会自动读了,需要手动通过 read 来读入。

  1. readableStream.readableFlowing=false;
  2.  
  3. letdata;
  4. while((data=readableStream.read())!=null){
  5. console.log(data.toString());
  6. }

但自己手动 read 比较麻烦,我们依然可以用自动流入的方式,调用 pause 和 resume 来暂停和恢复就行了。

当调用 writable stream 的 write 方法的时候会返回一个 boolean 值代表是写入了目标还是放在了缓冲区:

  • true: 数据已经写入目标
  • false:目标不可写入,暂时放在缓冲区

我们可以判断返回 false 的时候就 pause,然后等缓冲区清空了就 resume:

  1. constrs=fs.createReadStream(src);
  2. constws=fs.createWriteStream(dst);
  3.  
  4. rs.on('data',function(chunk){
  5. if(ws.write(chunk)===false){
  6. rs.pause();
  7. }
  8. });
  9.  
  10. rs.on('end',function(){
  11. ws.end();
  12. });
  13.  
  14. ws.on('drain',function(){
  15. rs.resume();
  16. });

这样就能达到根据写入速率暂停和恢复读入速率的功能,解决了背压问题。

pipe 有背压问题么?

平时我们经常会用 pipe 来直接把 Readable 流对接到 Writable 流,但是好像也没遇到过背压问题,其实是 pipe 内部已经做了读入速率的动态调节了。

  1. constrs=fs.createReadStream(src);
  2. constws=fs.createWriteStream(dst);
  3.  
  4. rs.pipe(ws);

总结

流是传输数据时常见的思想,就是一部分一部分的传输内容,是文件读写、网络通信的基础概念。

Node.js 也提供了 stream 的 api,包括 Readable 可读流、Writable 可写流、Duplex 双工流、Transform 转换流。它们分别实现 _read、_write、_read + _write、_transform 方法,来做数据的返回和处理。

创建 Readable 对象既可以直接调用 Readable api 创建,然后重写 _read 方法,也可以继承 Readable 实现一个子类,之后实例化。其他流同理。(Readable 可以很容易的和 generator 结合)

当读入的速率大于写入速率的时候就会出现“背压”现象,会爆缓冲区导致数据丢失,解决的方式是根据 write 的速率来动态 pause 和 resume 可读流的速率。pipe 就没有这个问题,因为内部做了处理。

流是掌握 IO 绕不过去的一个概念,而背压问题也是流很常见的问题,遇到了数据丢失可以考虑是否发生了背压。希望这篇文章能够帮大家理清思路,真正掌握 stream!

原文链接:https://mp.weixin.qq.com/s/9bf53AR2ephzcMTv5P5dJg