Node 之 stream

本篇对Node stream 进行深入了解，stream分为可写流和可读流。

stream

可写流事件

写入操作是异步的

1.可写流的基本例子

js
const fs=require('fs')
const file=fs.createWriteStream('a1.txt')
file.write('一些数据\n');
file.write('更多数据\n');
file.end('完成写入数据');

2.drain事件，监听当超出缓冲区后,何时缓冲区被释放

短时间内写入大量数据，超出缓冲区时（默认16kb），此时运行write会暂时无法写入。

使用drain可以监听当前写入是否超出缓冲区，作出相应处理

js
const fs=require('fs')
const file=fs.createWriteStream('a1.txt')

let i=100000
let ok=true
function write(){
  while(i>0&&ok){
    i--
    if(i===0){
      file.write('一些数据\n','utf-8',()=>{
        console.error('写入已完成');
      })
    }
    else{
      ok=file.write('一些数据\n','utf-8');
    }
  }
  if(i>0){
    ok=true
    console.log(`还剩下${i}`)
    file.once('drain', write);
  }
}
write()
// file.end('完成写入数据'); //不能在这里加入end，因为end会强制结束

该例子，所有数据都会全部显示。证明超出缓存区以后，数据依然会被写入（后面的800多条数据都会在缓冲区）

js
const file=fs.createWriteStream('a1.txt')
let i=0
let ok=true
function write(){
  for(;i<2000;i++){
    ok=file.write(`一些数据${i}\n`,'utf-8');
    if(!ok){
      file.once('drain', ()=>{
        console.log('缓冲区被占满')
      });
      console.log(i)
    }
  }
}
write()
// file.end('一些数据\n','utf-8'); 
// 添加该行，结果不会有任何改变，但是将无法看到 console.log('缓冲区被占满')，猜测是因为end强制输出。

3.finish事件，只有当触发end方法或者pipe方法完成传输时，才会触发finish事件

js
const fs=require('fs')
const file=fs.createWriteStream('a1.txt')
file.write('一些数据\n')
file.write('一些数据\n')
file.end('完成写入数据');
file.on('finish',()=>{
  console.error('写入已完成');
})

4.error事件，当发生报错时，会触发该事件。如果定义该事件，报错则不会退出进程，否则将退出进程

js
file.write('一些数据\n','utf-8');
file.destroy(new Error('12345'))

setInterval(()=>{
  console.log('进程正在进行') //维持进程
},500)

file.on('error',()=>{
  console.log('出现错误')
})

5.close事件，当关闭可读流时，会触发该事件。比如destory（不进行error报错），end等

js
file.write('一些数据\n','utf-8');
//file.destroy() //也会触发close事件

file.end('一些数据\n','utf-8');
file.on('close',()=>{
  console.log('进行已结束')
})

6.pipe事件，监听可写流是否通过pipe传输

js
const fs=require('fs')
const writer=fs.createWriteStream('write.txt')
const reader=fs.createReadStream('read.txt')

writer.on('pipe', (src) => {
  console.log('有数据正通过管道流入写入器');});
writer.on('drain',()=>{
  console.log('缓冲区已被释放')
})
reader.pipe(writer);

7.unpipe，监听可写流pipe传输是否被移除

可写流方法和属性

方法

writable.cork()

强制把所有写入的数据都缓冲到内存中。drain事件将会在最后被写入的时候触发一次。当调用 stream.uncork() 或 stream.end() 方法时，缓冲的数据才会被输出。

writable.uncork()

释放缓冲数据，输出到目标

建议使用 process.nextTick() 来延迟调用 writable.uncork()。通过这种方式，可以对单个 Node.js 事件循环中调用的所有 writable.write() 进行批处理。

js
stream.cork();
stream.write('一些 ');
stream.write('数据 ');
process.nextTick(() => stream.uncork());

writable.destroy([error])

销毁流。填写error则触发 'error'事件，不填写则触发 'close' 事件

writable.end([chunk[, encoding]][, callback])

触发finish事件和close事件。先触发finish事件，在触发close事件

writable.write(chunk[, encoding][, callback])

写入数据到流

writable.setDefaultEncoding(encoding)

方法为可写流设置默认的 encoding

属性

writable.destroyed：在调用了 writable.destroy() 之后为 true
writable.writable：如果调用 writable.write() 是安全的，则为 true
writable.writableEnded：在调用了 writable.end() 之后为 true。此属性不表明数据是否已刷新，对此请使用 writable.writableFinished
writable.writableFinished：在触发 'finish' 事件之立即设置为 true
writable.writableHighWaterMark：返回构造可写流时传入的 highWaterMark 的值
writable.writableLength：此属性包含准备写入的队列中的字节数（或对象）。该值提供有关 highWaterMark 状态的内省数据
writable.writableObjectMode：获取用于给定 Writable 流的 objectMode 属性

可读流介绍

流动模式（flowing）或暂停模式（paused）

在流动模式中，数据自动从底层系统读取，并通过 EventEmitter 接口的事件尽可能快地被提供给应用程序。
在暂停模式中，必须显式调用 stream.read() 读取数据块。

所有可读流都开始于暂停模式，可以通过以下方式切换到流动模式：

添加 'data' 事件句柄。
调用 stream.resume() 方法。
调用 stream.pipe() 方法将数据发送到可写流。

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

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

只有提供了消费或忽略数据的机制后，可读流才会产生数据。如果消费的机制被禁用或移除，则可读流会停止产生数据。

为了向后兼容，移除 'data' 事件句柄不会自动地暂停流。如果有管道目标，一旦目标变为 drain 状态并请求接收数据时，则调用 stream.pause() 也不能保证流会保持暂停模式。

如果可读流切换到流动模式，且没有可用的消费者来处理数据，则数据将会丢失。例如，当调用 readable.resume() 时，没有监听 'data' 事件或 'data' 事件句柄已移除。

添加 'readable' 事件句柄会使流自动停止流动，并通过 readable.read() 消费数据。如果 'readable' 事件句柄被移除，且存在 'data' 事件句柄，则流会再次开始流动。

可读流事件

data事件

使用该事件，流将会被转换到流动模式

data事件的优先级比readable优先级低，使用readable将被转换为暂停模式

当调用 readable.read() 且有数据块返回时，也会触发 'data' 事件

使用data事件，会自动读取文件

js
const fs = require('fs')
const reader = fs.createReadStream('read.txt')
reader.setEncoding('utf8')
reader.on('data', function (chunk) {
  console.log(chunk)
})

end事件

事件只有在数据被完全消费掉后才会触发（pipe，data，readable都可以触发）

close事件

pipe，data，readable都可以触发,destroy不包含error也可以触发

readable事件

监听数据变化，如果有新的数据，或者数据到达流的尽头,才会触发readable事件

js
const fs = require('fs')
const reader = fs.createReadStream('read.txt')

reader.setEncoding('utf8')
reader.on('readable', function (chunk) {
  let result=reader.read()
  console.log('chunk',result)
})
reader.on('end',function(){
  console.log('end')
})

error事件

监听报错，如果出现报错，则进入该事件

pause事件

监听pause方法调用，调用后执行该事件（流动模式下才可以暂停）

js
const fs = require('fs')
const reader = fs.createReadStream('read.txt')

reader.setEncoding('utf8')
let i=0
reader.on('data', function (chunk) {
  i++
  console.log('chunk')
  if(i>2){
    reader.pause()
  }
})
reader.on('pause',function(){
  console.log('pause')
})
reader.on('end',function(){
  console.log('end')
})

resume事件

监听resume方法调用

可读流方法和属性

方法

readable.destroy([error])

销毁流。填写error则触发 'error'事件，不填写则触发 'close' 事件

readable.isPaused()

方法返回可读流当前的操作状态。

readable.pause()

方法使流动模式的流停止触发 'data' 事件，并切换出流动模式。任何可用的数据都会保留在内部缓存中，readable事件下无法使用

readable.resume()

readable.resume() 方法将被暂停的可读流恢复触发 'data' 事件，并将流切换到流动模式，readable事件下无法使用

readable.pipe(destination[, options])

方法绑定可写流到可读流，将可读流自动切换到流动模式，并将可读流的所有数据推送到绑定的可写流。数据流会被自动管理，所以即使可读流更快，目标可写流也不会超负荷。

js
const fs = require('fs')
const reader = fs.createReadStream('read.txt')
const writer=fs.createWriteStream('write.txt')
reader.pipe(writer,{
  end:false //为false 的话，pipe write不会使用end写入
})
reader.on('end', () => {
  writer.end('结束');
});

readable.read([size])

可选的 size 参数指定要读取的特定字节数。如果无法读取 size 个字节，则除非流已结束，否则将会返回 null，在这种情况下，将会返回内部 buffer 中剩余的所有数据。

readable.setEncoding(encoding)

设置编码格式

readable.unpipe([destination])

移除管道流

属性

readable.destroyed：在调用了 readable.destroy() 之后为 true
readable.readable:如果可以安全地调用 readable.read()，则为 true
readable.readableEncoding:获取用于给定可读流的 encoding 属性。可以使用 readable.setEncoding() 方法设置 encoding 属性
readable.readableEnded:当 'end' 事件被触发时变为 true
readable.readableHighWaterMark:返回构造可读流时传入的 highWaterMark 的值

目录