Node.js 流 速查表

本文根据 Devhints / Node.js streams 翻译。代码示例及版本范围沿用原文。原文标注更新日期：2017-08-30。

类型

参见： 流 (nodejs.org)

流

const Readable = require('stream').Readable
const Writable = require('stream').Writable
const Transform = require('stream').Transform

管道 (Piping)

clock()              // Readable stream
  .pipe(xformer())   // Transform stream
  .pipe(renderer())  // Writable stream

方法

stream.push(/*...*/)         // Emit a chunk
stream.emit('error', error)  // Raise an error
stream.push(null)            // Close a stream

事件

const st = source()
st.on('data', (data) => { console.log('<-', data) })
st.on('error', (err) => { console.log('!', err.message) })
st.on('close', () => { console.log('** bye') })
st.on('finish', () => { console.log('** bye') })

假设 source() 是一个可读流。

流动模式 (Flowing mode)

// Toggle flowing mode
st.resume()
st.pause()

// Automatically turns on flowing mode
st.on('data', /*...*/)

流的类型

可读 (Readable)

function clock () {
  const stream = new Readable({
    objectMode: true,
    read() {}
  })

  setInterval(() => {
    stream.push({ time: new Date() })
  }, 1000)

  return stream
}

// Implement read() if you
// need on-demand reading.

可读流是数据生成器。使用……写入数据 stream.push().

转换 (Transform)

function xformer () {
  let count = 0

  return new Transform({
    objectMode: true,
    transform: (data, _, done) => {
      done(null, { ...data, index: count++ })
    }
  })
}

将更新后的数据块 Pass 传给 done(null, chunk).

可写 (Writable)

function renderer () {
  return new Writable({
    objectMode: true,
    write: (data, _, done) => {
      console.log('<-', data)
      done()
    }
  })
}

综合示例

clock()              // Readable stream
  .pipe(xformer())   // Transform stream
  .pipe(renderer())  // Writable stream

另请参阅

• https://nodejs.org/api/stream.html
• https://github.com/substack/stream-handbook