stream

[Genluo]

stream

Stream 是一个抽象接口，Node 中有很多对象实现了这个接口。例如，对http 服务器发起请求的request 对象就是一个 Stream，还有stdout（标准输出）

文档部分

一、基本的流类型：

• 可写入流(Writable)
• 可读取流(Readable)
• 可读又可写流(Duplex)
• 转化流(Transform)

二、类型

• 对象模式

Node.js 创建的流都是运作在字符串和 Buffer（或 Uint8Array）上。 当然，流的实现也可以使用其它类型的 JavaScript 值（除了 null）。 这些流会以“对象模式”进行操作。

• 缓冲

可写流和可读流都会在内部的缓冲器中缓存数据，可以分别使用的 writable.writableBuffer 或 readable.readableBuffer 来获取Buffer对象。可缓冲大大小由highWaterMark参数来确定

// 可读缓冲

一旦内部的可读缓冲的总大小达到 highWaterMark 指定的阈值时，流会暂时停止从底层资源读取数据，直到当前缓冲的数据被消费

// 可写缓冲

当内部的可写缓冲的总大小小于 highWaterMark 设置的阈值时，调用 writable.write() 会返回 true。 一旦内部缓冲的大小达到或超过 highWaterMark 时，则会返回 false

// 可读又可写缓冲

因为 Duplex 和 Transform 都是可读又可写的，所以它们各自维护着两个相互独立的内部缓冲器用于读取和写入， 这使得它们在维护数据流时，读取和写入两边可以各自独立地运作。

三、API

(1) 可写流

• 事件
  • close事件 ： 当流或者底层资源被关闭时触发，表明不会触发其他事件，也不会在发生操作，但不是每一个可写流都存在close事件
  • drain事件： 当可写流的buffer缓冲器满了调用，也就是write返回true的时候
  • error事件： 当写入数据发生错误的时候调用
  • finish事件： 调用end方法，并且缓冲数据已经传给底层系统之后触发
  • pipe事件： 当在可写流上调用pipe方法调用
  • unpipe事件： 当可读流通过管道流向可写流发生错误时
• 方法
  • cork()：将写入的数据都缓冲到内存中，当调用 stream.uncork() 或 stream.end() 时，缓冲的数据才会被输出，并且是一一对应的
  • uncork(): 将调用cork之后的缓存在内存的数据输出到目标中，如果一个流上调用同样次数的writable.uncork() 才能输出缓冲的数据
  • destroy()：销毁流，
  • end()：将缓冲区的数据最后一次强制写入底层资源中
  • setDefaultEncoding(): 为可写流设置编码
  • write：(第一次真的往文件中填写，后面写到缓冲中buffer) 这里会产生一个：背压问题
• 属性
  • readableHighWaterMark
  • Writable
  • WritableHighWaterMark
  • writableLength 队列中准备写入的字节数

背压问题：

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

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

(2) 可读流

• 读取模式

  • 流动模式

  数据自动从底层系统读取，并通过 EventEmitter 接口的事件尽可能快地被提供给应用程序。

  • 暂停模式

  必须显式调用 stream.read() 读取数据块。

  两种模式的切换： 所有的可读流刚开始都处于暂停模式，可以通过以下方法切换为流动模式

  • 添加 'data' 事件句柄。
  • 调用 stream.resume()。
  • 调用 stream.pipe()。

  可读流可以通过以下方式切换为暂停模式

  • 如果没有管道目标，则调用 stream.pause()。
  • 如果有管道目标，则移除所有管道目标。调用 stream.unpipe() 可以移除多个管道目标。

• 事件
  • close ： 当流或其底层资源（比如文件描述符）被关闭时触发。 表明不会再触发其他事件，也不会再发生操作。
  • data： 当流将数据块传送给消费者后触发
  • end： 当流中没有数据可供消费时触发
  • error： 当流因底层内部出错而不能产生数据、或推送无效的数据块时触发。
  • readable： 当流中有数据可供读取时触发，当到达流数据的尽头时，readable事件也可以触发，但是在end事件之前触发
• 方法
  • destroy：销毁流并触发error事件和close事件
  • isPaused： 返回可读流当前的操作状态
  • pause： 使得流动模式的流触发data事件，并切换出流动模式，任何可用的数据都会保留在内部缓存中。
  • pipe： 绑定可写流到可读流，将可读流自动切换到流动模式，并将可读流的所有数据推送到绑定的可写流。 数据流会被自动管理，所以即使可读流更快，目标可写流也不会超负荷。
  • read： 从内部缓冲拉取并返回数据，通常是Buffer对象
  • readsume()： 将暂停的可读流恢复触发data事件，并将流切换到流动模式。
  • setEncoding()：设置可度流的编码，默认情况下没有设置编码，流数据返回的是buffer对象，如果设置了编码，那么将返回指定编码的字符串
  • unpipe：移除管道的可写流，将可读流的模式转化为暂停模式
  • unshift: 将数据块推回内部缓冲。 可用于以下情景：正被消费中的流需要将一些已经被拉出的数据重置为未消费状态，以便这些数据可以传给其他方。
• 属性
  • readable: 是否安全调用read()方法
  • readableHighWaterMark： 返回构造时候传入的highWaterMark
  • readableLength： 队列中准备读取的字节数

(3) 双向流和转化流

(1) 简单实现

对于简单的案例，构造流可以不依赖继承。 直接创建 stream.Writable、 stream.Readable、 stream.Duplex 或 stream.Transform 的实例，并传入对应的方法作为构造函数选项。

const { Writable } = require('stream');

const myWritable = new Writable({
  write(chunk, encoding, callback) {
    // ...
  },
  writev(chunks, callback) {
    // ...
  }
});

(2) 类实现

• 可写流实现

const { Writable } = require('stream');

class MyWritable extends Writable {
  constructor(options) {
    // 调用 stream.Writable() 构造函数。
    super(options);
    // ...
  }
}

(4) 类静态方法

• finised： 当流不再可读、可写、发生错误、或提前关闭时，通过该函数获得通知。
• pipeline: 用管道连接两个或者多个流

(5) 实现流接口

理解部分

一、为什么要使用stream

• 处理数据比较及时，导致处理数据比较快

• 节省空间

• 可以随意结合使用，扩展性高

二、流的实质

stream = Buffer + Events

我么可以这样理解node中流，就是针对相关读写操作使用buffer进行缓存，然后通过事件进行通信，所以整体可以理解为stream = Buffer + EventEmitter

三、如何解决写入时的背压问题

(1) 使用pipe进行流传递，这是一种流模式，可以自动处理背压问题

• 不断从来源可读流中获得一个指定长度的数据
• 将获取到的数据写入到目标可写流中
• 平衡读取和写入的速度，防止读取速度大大超出写入速度，出现大量滞留数据

(2) 我们手动来判断是否产生背压问题

• write返回的为true
• 停止写入数据

三、简单实现流

可读流的实现：

let EventEmitter = require('events');
let fs = require('fs');

class ReadStream extends EventEmitter {
  constructor(path, options) {
    super();
    this.path = path;
    this.autoClose = options.autoClose || true;
    this.flags = options.flags || 'r';
    this.encoding = options.encoding || null;
    this.start = options.start || 0;
    this.end = options.end || null;
    this.highWaterMark = options.highWaterMark || 64 * 1024;
    // 应该有一个读取文件的位置 可变的（可变的位置）
    this.pos = this.start;
    // 控制当前是否是流动模式
    this.flowing = null;
    // 构建读取到的内容的buffer
    this.buffer = Buffer.alloc(this.highWaterMark);
    // 但创建可读流 要将文件打开
    this.open();
    this.on('newListener', (type) => {
      if (type === 'data') {
        // 如果用户监听了data事件，就开始读取
        this.flowing = true;
        this.read(); // 开始读取文件
      }
    })
  }
  read() {
    // 这时候文件还没有打开，等文件打开后再去读取
    if (typeof this.fd !== 'number') {
      // 等待文件打开，再次调用read方法
      return this.once('open', () => this.read())
    }
    // 开始读取
    let howMuchToRead = this.end ? Math.min(this.highWaterMark, this.end - this.pos + 1) : this.highWaterMark;
    // 文件描述符 读到哪个buffer里  读取到buffer的哪个位置 往buffer里读取几个 读取的位置
    // 想读三个 但是文件只有2个
    // todo 外部调用resume方法的时候 需判断如果end了 就不要读了
    fs.read(this.fd, this.buffer, 0, howMuchToRead, this.pos, (err, bytesRead) => {
      if (err) {
        this.emit('error')
        return
      }
      if (bytesRead > 0) { // 读到内容了
        this.pos += bytesRead;
        // 保留有用的
        let r = this.buffer.slice(0, bytesRead);
        r = this.encoding ? r.toString(this.encoding) : r;
        this.emit('data', r);
        if (this.flowing) {
          this.read();
        }
      } else {
        this.emit('end');
        this.destroy();
      }
    });
  }
  destroy() {
    if (typeof this.fd === 'number') {
      fs.close(this.fd, () => {
        this.emit('close');
      });
      return
    }
    this.emit('close');
  }
  open() { // 打开文件的逻辑
    fs.open(this.path, this.flags, (err, fd) => {
      if (err) {
        this.emit('error', err);
        if (this.autoClose) {
          this.destroy() // 销毁  关闭文件 （触发close事件）
        }
        return
      }
      this.fd = fd;
      this.emit('open'); // 触发文件开启事件
    })
  }
  pause() {
    this.flowing = false;
  }
  resume() {
    this.flowing = true;
    this.read();
  }
}

可写流实现：

let fs = require('fs');
let EventEmitter = require('events');
class WriteStream extends EventEmitter {
    constructor(path, options) {
        super();
        this.path = path;
        this.flags = options.flags || 'w';
        this.mode = options.mode || 0o666;
        this.highWaterMark = options.highWaterMark || 16 * 1024;
        this.start = options.start || 0;
        this.autoClose = options.autoClose || true;
        this.encoding = options.encoding || 'utf8';
        // 是否需要触发drain事件
        this.needDrain = false;
        // 是否正在写入
        this.writing = false;
        // 缓存 正在写入就放到缓存中
        this.buffer = [];
        // 算一个当前缓存的个数
        this.len = 0;
        // 写入 的时候也有位置关系
        this.pos = this.start;
        this.open();
    }
    open() {
        fs.open(this.path, this.flags, this.mode, (err, fd) => {
            if (err) {
                this.emit('error');
                this.destroy();
                return
            }
            this.fd = fd;
            this.emit('open');
        })
    }
    destroy() {
        if (typeof this.fd === 'number') {
            fs.close(this.fd, () => {
                this.emit('close');
            });
            return
        }
        this.emit('close');
    }
    write(chunk, encoding = this.encoding, callback) {
        chunk = Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk);
        this.len += chunk.length;
        this.needDrain = this.highWaterMark <= this.len;
        if (this.writing) {
            this.buffer.push({chunk, encoding, callback})
        } else {
            // 当文件写入沟 清空缓存区的内容
            this.writing = true; // 走缓存
            this._write(chunk, encoding, () => this.clearBuffer())
        }
        return !this.needDrain;
    }
    _write(chunk, encoding, callback) {
        if (typeof this.fd !== 'number') {
            return this.once('open', () => this.write(chunk, encoding, callback));
        }
        // fd是文件描述符 chunk是数据  0 是写入的位置  写入的长度， this.pos 偏移量
        fs.write(this.fd, chunk, 0, chunk.length, this.pos, (err, bytesWritten) => {
            this.pos += bytesWritten;
            this.len -= bytesWritten; // 写入的长度会减少
            callback();
        })
    }
    clearBuffer() {
        let buf = this.buffer.shift();
        if(buf) {
            this._write(buf.chunk, buf.encoding, () => this.clearBuffer());
        } else {
            this.writing = false;
            this.needDrain = false;
            this.emit('drain')
        }
    }
}
module.exports = WriteStream;

安利一stream源码实现流程图，虽然自己没有去看过：

参考资料

• 对Node.js中 stream模块的学习积累和理解

[Genluo]

深度学习好文推荐： https://blog.5udou.cn/blog/NodejsLiu-Xue-Xi-Xi-Lie-Zhi-Yi-Readable-Stream31 https://blog.5udou.cn/blog/NodejsLiu-Xue-Xi-Xi-Lie-Zhi-Er-Writable-Stream91 https://blog.5udou.cn/blog/NodejsLiu-Xue-Xi-Xi-Lie-Zhi-San-Duplex-Stream-Transform-Stream42