Stream（流）的基本使用

流 stream 是一组有序的，有起点和终点的字节数据传输手段，而且有不错的效率。 借助事件和非阻塞I/O库,流模块允许在其可用的时候动态处理,在其不需要的时候释放掉。

流 stream 是一种在Node.js中处理流式数据的抽象接口。 stream 模块提供了以下基础的API，用于构建实现了流接口对象。 流可以是可读、可写、或是可读写的。

基本使用

可读流

在NodeJS中，我们对文件的操作需要依赖核心模块 fs ， fs 模块中集成了 createReadStream 可读流。

fs.createReadStream(path, options) 参数如下：

• path : 读取的文件路径

• options : string(指定字符编码) | Object(下面详细介绍)

  flags
  encoding
  fd
  mode
  autoClose
  start
  end
  highWaterMark
  

• 返回值为： fs.ReadStream

创建可读流

// 引入依赖
let fs = require('fs')
let rs = fs.createReadStream('./readStream.txt', {
  // start: 0, // 开始读取位置，默认0
  end: 3, // 结束位置，默认文章读取完
  highWaterMark: 3, // 最多读取， 每次读取的个数 默认：64*1024 字节
})
复制代码

可读流中的事件机制

在下面例子中 ./readStream.txt 的文件内容为: 1234567890

1. open

open 事件用来监听文件的打开，回调函数在打开文件后执行。

// 引入依赖
let fs = require('fs')
let rs = fs.createReadStream('./readStream.txt', {
  start: 0,
  end: 3,
  highWaterMark: 3
})
// 事件机制，需要自己去监听一些数据
// open 文件打开
rs.on('open', () => {
  console.log('文件开启了')
})
// 结果：文件开启了
复制代码

2. data

data 事件，每次读取 highWaterMark 个字节，就触发一次该事件，直到读取完成，回调函数里面返回的时每次读取的结果。如果 encoding 不设置，则返回为Buffer

// 引入依赖
let fs = require('fs')
let rs = fs.createReadStream('./readStream.txt', {
  // encoding: 'utf8',
  start: 0,
  end: 3,
  highWaterMark: 3
})
// 事件机制，需要自己去监听一些数据
// open 文件打开
rs.on('open', () => {
  console.log('文件开启了')
})
rs.on('data', (data) => {
  console.log(data)
})
// 如果不是设置encoding返回结果为
/** 文件开启了
 *<Buffer 31 32 33>
 * <Buffer 34>
 */
 
// 如果encoding: 'utf8'返回结果为
/** 文件开启了
 * 123
 * 4
 */
复制代码

3. end

end 事件，当文件读取完触发，并执行回调。

读取完情况如下：

1. end 设置了值（ end: 3 ），则读完设置的长度触发
2. end 没有设置值，默认读取完所有内容触发。

为了方便查看结果，在下面例子中，我们也实现了这两种情况：

// 引入依赖
let fs = require('fs')
let rs = fs.createReadStream('./readStream.txt', {
  start: 0,
  // end: 3,
  highWaterMark: 3
})
// 事件机制，需要自己去监听一些数据
// open 文件打开
rs.on('open', () => {
  console.log('文件开启了')
})
rs.on('data', (data) => {
  console.log(data)
})
rs.on('end', () => {
  console.log('结束了')
})
// 设置了end: 3的运行结果
/** 文件开启了
 * <Buffer 31 32 33>
 * <Buffer 34>
 * 结束了
 */
 
// 没有设置end参数的运行结果
/** 文件开启了
 * <Buffer 31 32 33>
 * <Buffer 34 35 36>
 * <Buffer 37 38 39>
 * <Buffer 30>
 * 结束了
 */
复制代码

4. error

error 事件监听错误信息，在读文件出差时触发回调，并将错误信息返回

// error事件 出错时自动调用
rs.on('error', (err) => {
  console.log(err)
})
复制代码

5. close

close 事件用来监听文件的关闭，回调函数在文件关闭后执行。在创建可读流时， autoClose: true （默认值为true），自动关闭文件，且触发 close 事件执行回调。

rs.on('close', () => {
  console.log('close')
})
复制代码

暂停和恢复

可读流有两种状态：流动状态或暂停状态两种。

可读流开始都时暂停状态，可以通过以下方式切换到流动状态：

data
fs.resume()

可读流可以通过以下方式切换回暂停状态：

• 调用 rs.pause() 方法

什么情况先我们会用到暂停和恢复呢？

我们都知道读取文件会占用内容空间，如果我们遇到特别大的文件读取时，如果全部读出会占用很大的内容空间，这不是我们想要看到的，我们想到读取一部分数据，处理完成后再次读取，就会用到了。

一个简单的例子

// 引入依赖
let fs = require('fs')
let rs = fs.createReadStream('./readStream.txt', {
  encoding: 'utf8', // 字符编码, 默认为null
  highWaterMark: 2
})
let i = 0

rs.on('data', (data) => {
  i ++
  console.log(`第 ${i} 次`, new Date());
  console.log(data)
  rs.pause() // 暂停
  setTimeout(() => {
    rs.resume() // 恢复
  }, 1000)
})
rs.on('end', () => {
  console.log('结束了')
})
// 第 1 次 2018-09-16T10:11:00.993Z
// 12
// 第 2 次 2018-09-16T10:11:01.996Z
// 34
// 第 3 次 2018-09-16T10:11:02.997Z
// 56
// 第 4 次 2018-09-16T10:11:03.997Z
// 78
// 第 5 次 2018-09-16T10:11:04.998Z
// 90
// 结束了
复制代码

可写流

在NodeJS中，我们对文件的操作需要依赖核心模块 fs ， fs 模块中集成了 createWriteStream 可读流

fs.createWriteStream(path, options) 参数如下：

• path : 读取的文件路径

• options : string(指定字符编码) | Object

  • flags : 标识位，默认为 'w'
  • encoding : 字符编码, 默认为 utf8
  • fd ：文件描述符，默认为 null；
  • mode ：权限位，默认为 0o666；
  • autoClose : 是否自动关闭，默认为 true
  • start : 开始读取位置，默认 0
  • highWaterMark : 写入个数

• 返回值为： fs.WriteStream

创建可写流

writeStream.txt 文件，表示要写入内容的文件

// 引入依赖
let fs = require('fs')
let ws = fs.createWriteStream('./writeStream.txt', {
  start: 0, // 开始读取位置，默认0
})
复制代码

写内容和结束

write 在对应的文件写入内容。 end 表示写入结束，如果后面参数有内容，也是可以入去到文件。

// 引入依赖
let fs = require('fs')
let ws = fs.createWriteStream('./writeStream.txt', {
  start: 0, // 开始写入位置，默认0
})
ws.write('1') // 写内容
ws.end('写完了') // 结束

//结果为： 1写完了
复制代码

可写流中的事件机制

可写流里面的事件 open 、 close 相对对简单不再详述。主要说一下 finish 和 drain 事件.

1. finish 事件

当调用 ws.end() 方法且缓冲数据都已经传给底层系统之后，触发 'finish' 事件。

测试代码如下：

// 引入依赖
let fs = require('fs')
let ws = fs.createWriteStream('./writeStream.txt', {
  start: 0, // 开始写入位置，默认0
})
ws.on('open', () => {
  console.log('open');
});
ws.write('1')
ws.end('写完了')
ws.on('finish', () => {
  console.log('所有写入已完成。');
});
ws.on('close', () => {
  console.error('close');
});

// 输出结果为：
// open
// 所有写入已完成。
// close
复制代码

2. drain 事件

如果调用 ws.write(chunk) 方法返回 false ，也就是写入的内容到达缓存区的大小，触发 drain 事件

let fs = require('fs')
let ws = fs.createdWriteStream('writeStream.txt', {
    start: 0,
    highWaterMark: 5
})

// 写入15个数，每次写五个，只希望占用五个字节的内存
let i = 0
function write() {
    let flag = true
    while(i < 15 && flag) {
        flag = ws.write(i++'','utf8')
    }
}
write() // 如果写入的内容到达缓存区的大小了，当他写入完成后会触发一个事件
ws.on('drain', () => {
    console.log('占满')
    write() //清空缓存 继续写入
})

// 占满
// 占满
复制代码

可读可写混合使用

通过 pipe 方式读取一个文件内容，写入到另外一个文件【常用】

// 引入fs模块
const fs = require("fs");

// 创建可读流和可写流
let rs = fs.createReadStream("./ReadStream/readStream.txt", {
    highWaterMark: 3
});
let ws = fs.createWriteStream("./writeStream.txt", {
    highWaterMark: 2
});

// 将 readStream.txt 的内容通过流写入 writeStream.txt 中
rs.pipe(ws);
复制代码

附源码

重要 ： 为了方便大家了解、查看、调试代码，完整的源码参见 gitHub

总结

本篇文章是 NodeJs 中流（stream）的基础了解。希望对大家了解流起到一定的作用。

下篇 ： NodeJS 流的实现原理和简单实现—— 《流之原，源之理》