135-1821-9792

彻底掌握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)。

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

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

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

流的 api

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

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

它们都有要实现的方法：

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

我们分别来看一下：

Readable

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

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

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

执行效果如下：

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

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

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

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

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

Writable

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

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

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

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

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

Duplex

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

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

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

Transform

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

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

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

跑了一下，效果如下：

流的暂停和流动

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

背压

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

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

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

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

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

解决背压

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

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

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

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

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

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

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

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

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

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

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

pipe 有背压问题么?

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

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

总结

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

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!


网页标题：彻底掌握Node.js四大流，解决爆缓冲区的“背压”问题
URL地址：http://www.wtcwzsj.com/article/coegscs.html

其他资讯


关于我们
我们的服务
网站建设案例
新闻动态
联系我们

135-1821-9792

公司服务热线
友情链接： 成都企业网站建设   消防维保   成都大型发电机组出租   香港云主机   成都响应式网站建设   ptjierui.cn   四川发电机回收   成都导视系统设计   传美奇电气   app软件开发   

Copyright © 2009-2022 www.wtcwzsj.com 青羊区广皓图文设计工作室（个体工商户） 版权所有　蜀ICP备19037934号