Streams 是 Node.js 的組件和模式中最重要的幾個(gè)之一骡显。在 Node.js 這類基于 event 的平臺(tái)上，最高效的實(shí)時(shí)地處理 I/O 的方式梭稚，就是當(dāng)有輸入時(shí)就立即接收數(shù)據(jù)，應(yīng)用產(chǎn)生輸出時(shí)就立即發(fā)送數(shù)據(jù)。

Buffering vs streaming

對(duì)于輸入數(shù)據(jù)的處理哩照，buffer 模式會(huì)將來自資源的所有數(shù)據(jù)收集到 buffer 中，待操作完成再將數(shù)據(jù)作為單一的 blob of data 傳遞給調(diào)用者懒浮；相反地飘弧，streams 允許我們一旦接收到數(shù)據(jù)就立即對(duì)其進(jìn)行處理。
單從效率上說砚著，streams 在空間（內(nèi)存使用）和時(shí)間（CPU 時(shí)鐘）的使用上都更加高效次伶。此外 Node.js 中的 streams 還有另一個(gè)重要的優(yōu)勢：組合性。

空間效率

使用 buffered API 完成 Gzip 壓縮：

import {promises as fs} from 'fs'
import {gzip} from 'zlib'
import {promisify} from 'util'

const gzipPromise = promisify(gzip)
const filename = process.argv[2]

async function main() {
  const data = await fs.readFile(filename)
  const gzippedData = await gzipPromise(data)
  await fs.writeFile(`${filename}.gz`, gzippedData)
  console.log('File successfully compressed')
}

main()

node gzip-buffer.js <path to file>

如果我們使用上述代碼壓縮一個(gè)足夠大的文件（比如說 8G）稽穆，我們很有可能會(huì)收到一個(gè)錯(cuò)誤信息冠王，類似文件大小超過了允許的最大 buffer 大小。

RangeError [ERR_FS_FILE_TOO_LARGE]: File size (8130792448) is greater
than possible Buffer: 2147483647 bytes

即便沒有超過 V8 的 buffer 大小限制秧骑，也有可能出現(xiàn)物理內(nèi)存不夠用的情況版确。

使用 streams 實(shí)現(xiàn) Gzip 壓縮：

import {createReadStream, createWriteStream} from 'fs'
import {createGzip} from 'zlib'

const filename = process.argv[2]

createReadStream(filename)
  .pipe(createGzip())
  .pipe(createWriteStream(`${filename}.gz`))
  .on('finish', () => console.log('File successfully compressed'))

streams 的優(yōu)勢來自于其接口和可組合性扣囊，允許我們實(shí)現(xiàn)干凈、優(yōu)雅绒疗、簡潔的代碼侵歇。對(duì)于此處的示例，它可以對(duì)任意大小的文件進(jìn)行壓縮吓蘑，只需要消耗常量的內(nèi)存惕虑。

時(shí)間效率

假設(shè)我們需要?jiǎng)?chuàng)建一個(gè)應(yīng)用，能夠壓縮一個(gè)文件并將其上傳到一個(gè)遠(yuǎn)程的 HTTP 服務(wù)器磨镶。而服務(wù)器端則負(fù)責(zé)將接收到的文件解壓縮并保存溃蔫。
如果我們使用 buffer API 實(shí)現(xiàn)客戶端組件，則只有當(dāng)整個(gè)文件讀取和壓縮完成之后琳猫，上傳操作才開始觸發(fā)伟叛。同時(shí)在服務(wù)器端，也只有當(dāng)所有數(shù)據(jù)都接收完畢之后才開始解壓縮操作脐嫂。

更好一些的方案是使用 streams统刮。在客戶端，streams 允許我們以 chunk 為單位從文件系統(tǒng)逐個(gè)账千、分段地讀取數(shù)據(jù)侥蒙，并立即進(jìn)行壓縮和發(fā)送。同時(shí)在服務(wù)器端匀奏，每個(gè) chunk 被接收到后會(huì)立即進(jìn)行解壓縮鞭衩。

服務(wù)端程序：

import {createServer} from 'http'
import {createWriteStream} from 'fs'
import {createGunzip} from 'zlib'
import {basename, join} from 'path'

const server = createServer((req, res) => {
  const filename = basename(req.headers['x-filename'])
  const destFilename = join('received_files', filename)
  console.log(`File request received: ${filename}`)
  req
    .pipe(createGunzip())
    .pipe(createWriteStream(destFilename))
    .on('finish', () => {
      res.writeHead(201, {'Content-Type': 'text/plain'})
      res.end('OK\n')
      console.log(`File saved: ${destFilename}`)
    })
})

server.listen(3000, () => console.log('Listening on http://localhost:3000'))

客戶端程序：

import {request} from 'http'
import {createGzip} from 'zlib'
import {createReadStream} from 'fs'
import {basename} from 'path'

const filename = process.argv[2]
const serverHost = process.argv[3]

const httpRequestOptions = {
  hostname: serverHost,
  port: 3000,
  path: '/',
  method: 'PUT',
  headers: {
    'Content-Type': 'application/octet-stream',
    'Content-Encoding': 'gzip',
    'X-Filename': basename(filename)
  }
}

const req = request(httpRequestOptions, (res) => {
  console.log(`Server response: ${res.statusCode}`)
})

createReadStream(filename)
  .pipe(createGzip())
  .pipe(req)
  .on('finish', () => {
    console.log('File successfully sent')
  })

mkdir received_files
node gzip-receive.js
node gzip-send.js <path to file> localhost

借助 streams，整套流程的流水線在我們接收到第一個(gè)數(shù)據(jù)塊的時(shí)候就開始啟動(dòng)了娃善，完全不需要等待整個(gè)文件被讀取论衍。除此之外，下一個(gè)數(shù)據(jù)塊能夠被讀取時(shí)会放，不需要等到之前的任務(wù)完成就能被處理饲齐。即另一條流水線被并行地被裝配執(zhí)行，Node.js 可以將這些異步的任務(wù)并行化地執(zhí)行咧最。只需要保證數(shù)據(jù)塊最終的順序是固定的，而 Node.js 中 streams 的內(nèi)部實(shí)現(xiàn)機(jī)制保證了這一點(diǎn)御雕。

組合性

借助于 pipe() 方法矢沿，不同的 stream 能夠被組合在一起。每個(gè)處理單元負(fù)責(zé)各自的單一功能酸纲，最終被 pipe() 連接起來捣鲸。因?yàn)?streams 擁有統(tǒng)一的接口，它們彼此之間在 API 層面是互通的闽坡。只需要 pipeline 支持前一個(gè) stream 生成的數(shù)據(jù)類型（可以是二進(jìn)制栽惶、純文本甚至對(duì)象等）愁溜。

客戶端加密

import {createCipheriv, randomBytes} from 'crypto'
import {request} from 'http'
import {createGzip} from 'zlib'
import {createReadStream} from 'fs'
import {basename} from 'path'

const filename = process.argv[2]
const serverHost = process.argv[3]
const secret = Buffer.from(process.argv[4], 'hex')
const iv = randomBytes(16)

const httpRequestOptions = {
  hostname: serverHost,
  port: 3000,
  path: '/',
  method: 'PUT',
  headers: {
    'Content-Type': 'application/octet-stream',
    'Content-Encoding': 'gzip',
    'X-Filename': basename(filename),
    'X-Initialization-Vector': iv.toString('hex')
  }
}

const req = request(httpRequestOptions, (res) => {
  console.log(`Server response: ${res.statusCode}`)
})

createReadStream(filename)
  .pipe(createGzip())
  .pipe(createCipheriv('aes192', secret, iv))
  .pipe(req)
  .on('finish', () => {
    console.log('File successfully sent')
  })

服務(wù)端加密

import {createServer} from 'http'
import {createWriteStream} from 'fs'
import {createGunzip} from 'zlib'
import {basename, join} from 'path'
import {createDecipheriv, randomBytes} from 'crypto'

const secret = randomBytes(24)
console.log(`Generated secret: ${secret.toString('hex')}`)

const server = createServer((req, res) => {
  const filename = basename(req.headers['x-filename'])
  const iv = Buffer.from(
    req.headers['x-initialization-vector'], 'hex'
  )
  const destFilename = join('received_files', filename)
  console.log(`File request received: ${filename}`)
  req
    .pipe(createDecipheriv('aes192', secret, iv))
    .pipe(createGunzip())
    .pipe(createWriteStream(destFilename))
    .on('finish', () => {
      res.writeHead(201, {'Content-Type': 'text/plain'})
      res.end('OK\n')
      console.log(`File saved: ${destFilename}`)
    })
})

server.listen(3000, () => console.log('Listening on http://localhost:3000'))

Streams 詳解

實(shí)際上在 Node.js 中的任何地方都可見到 streams。比如核心模塊 fs 有 createReadStream() 方法用來讀取文件內(nèi)容外厂，createWriteStream() 方法用來向文件寫入數(shù)據(jù)冕象；HTTP request 和 response 對(duì)象本質(zhì)上也是 stream；zlib 模塊允許我們通過流接口壓縮和解壓縮數(shù)據(jù)汁蝶；甚至 crypto 模塊也提供了一些有用的流函數(shù)比如 createCipheriv 和 createDecipheriv渐扮。

streams 的結(jié)構(gòu)

Node.js 中的每一個(gè) stream 對(duì)象，都是對(duì)以下四種虛擬基類里任意一種的實(shí)現(xiàn)掖棉，這四個(gè)虛擬類都屬于 stream 核心模塊：

• Readable
• Writable
• Duplex
• Transform

每一個(gè) stream 類同時(shí)也是 EventEmitter 的實(shí)例墓律，實(shí)際上 Streams 可以生成幾種類型的 event。比如當(dāng)一個(gè) Readable 流讀取完畢時(shí)觸發(fā) end 事件幔亥，Writable 流吸入完畢時(shí)觸發(fā) finish 事件耻讽，或者當(dāng)任意錯(cuò)誤發(fā)生時(shí)拋出 error。

Steams 之所以足夠靈活帕棉，一個(gè)重要的原因就是它們不僅僅能夠處理 binary data齐饮，還支持幾乎任意的 JavaScript 值。實(shí)際上 streams 有以下兩種操作模式：

• Binary mode：以 chunk 的形式（比如 buffers 或 strings）傳輸數(shù)據(jù)
• Object mode：通過由獨(dú)立對(duì)象（可以包含任意 JavaScript 值）組成的序列傳輸數(shù)據(jù)

上述兩種模式使得我們不僅僅可以利用 streams 處理 I/O 操作笤昨，還能夠幫助我們以函數(shù)式的方式將多個(gè)處理單元優(yōu)雅地組合起來祖驱。

從 Readable streams 讀取數(shù)據(jù)

non-flowing mode

默認(rèn)模式。readable 事件表示有新的數(shù)據(jù)可供讀取瞒窒，再通過 read() 方法同步地從內(nèi)部 buffer 讀取數(shù)據(jù)捺僻，返回一個(gè) Buffer 對(duì)象。
即從 stream 按需拉取數(shù)據(jù)崇裁。當(dāng) stream 以 Binary 模式工作時(shí)匕坯，我們還可以給 read() 方法指定一個(gè) size 值，以讀取特定數(shù)量的數(shù)據(jù)拔稳。

process.stdin
  .on('readable', () => {
    let chunk
    console.log('New data available')
    while ((chunk = process.stdin.read()) !== null) {
      console.log(
        `Chunk read (${chunk.length} bytes): "${chunk.toString()}"`
      )
    }
  })
  .on('end', () => console.log('End of stream'))

flowing mode

此模式下葛峻，數(shù)據(jù)并不會(huì)像之前那樣通過 read() 方法拉取巴比，而是一旦有數(shù)據(jù)可用术奖，就主動(dòng)推送給 data 事件的 listener。flowing 模式對(duì)于數(shù)據(jù)流的控制轻绞，相對(duì)而言靈活性較低一些采记。
由于默認(rèn)是 non-flowing 模式，為了使用 flowing 模式政勃，需要綁定一個(gè) listener 給 data 事件或者顯式地調(diào)用 resume() 方法唧龄。調(diào)用 pause() 方法會(huì)導(dǎo)致 stream 暫時(shí)停止發(fā)送 data 事件，任何傳入的數(shù)據(jù)會(huì)先被緩存到內(nèi)部 buffer奸远。即 stream 又切換回 non-flowing 模式既棺。

process.stdin
  .on('readable', () => {
    let chunk
    console.log('New data available')
    while ((chunk = process.stdin.read()) !== null) {
      console.log(
        `Chunk read (${chunk.length} bytes): "${chunk.toString()}"`
      )
    }
  })
  .on('end', () => console.log('End of stream'))

Async iterators

Readable 流同時(shí)也是 async iterators讽挟。

async function main() {
  for await (const chunk of process.stdin) {
    console.log('New data available')
    console.log(
      `Chunk read (${chunk.length} bytes): "${chunk.toString()}"`
    )
  }
  console.log('End of stream')
}

main()

實(shí)現(xiàn) Readable streams

import {Readable} from 'stream'
import Chance from 'chance'

const chance = Chance()

export class RandomStream extends Readable {
  constructor(options) {
    super(options)
    this.emittedBytes = 0
  }

  _read(size) {
    const chunk = chance.string({length: size})
    this.push(chunk, 'utf8')
    this.emittedBytes += chunk.length
    if (chance.bool({likelihood: 5})) {
      this.push(null)
    }
  }
}

const randomStream = new RandomStream()
randomStream
  .on('data', (chunk) => {
    console.log(`Chunk received (${chunk.length} bytes): ${chunk.toString()}`)
  })

為了實(shí)現(xiàn)一個(gè)自定義的 Readable stream，首先必須創(chuàng)建一個(gè)新的類丸冕，該類繼承自 stream 模塊中的 Readable耽梅。其次新創(chuàng)建的類中必須包含 _read() 方法的實(shí)現(xiàn)。
上面代碼中的 _read() 方法做了以下幾件事：

• 借助第三方的 chance 模塊晨仑，生成一個(gè)長度為 size 的隨機(jī)字符串
• 通過 push() 方法將字符傳推送到內(nèi)部 buffer
• 依據(jù) 5% 的幾率自行終止褐墅，終止時(shí)推送 null 到內(nèi)部 buffer，作為 stream 的結(jié)束標(biāo)志

簡化版實(shí)現(xiàn)

import {Readable} from 'stream'
import Chance from 'chance'

const chance = new Chance()
let emittedBytes = 0

const randomStream = new Readable({
  read(size) {
    const chunk = chance.string({length: size})
    this.push(chunk, 'utf8')
    emittedBytes += chunk.length
    if (chance.bool({likelihood: 5})) {
      this.push(null)
    }
  }
})

randomStream
  .on('data', (chunk) => {
    console.log(`Chunk received (${chunk.length} bytes): ${chunk.toString()}`)
  })

從可迭代對(duì)象創(chuàng)建 Readable streams

Readable.from() 方法支持從數(shù)組或者其他可迭代對(duì)象（比如 generators, iterators, async iterators）創(chuàng)建 Readable streams洪己。

import {Readable} from 'stream'

const mountains = [
  {name: 'Everest', height: 8848},
  {name: 'K2', height: 8611},
  {name: 'Kangchenjunga', height: 8586},
  {name: 'Lhotse', height: 8516},
  {name: 'Makalu', height: 8481}
]

const mountainsStream = Readable.from(mountains)
mountainsStream.on('data', (mountain) => {
  console.log(`${mountain.name.padStart(14)}\t${mountain.height}m`)
})

Writable streams

向流寫入數(shù)據(jù)

write() 方法可以向 Writable stream 寫入數(shù)據(jù)妥凳。
writable.write(chunk, [encoding], [callback])

end() 方法可以向 stream 表明沒有更多的數(shù)據(jù)需要寫入攒庵。
writable.end([chunk], [encoding], [callback])

callback 回調(diào)函數(shù)等同于為 finish 事件注冊(cè)了一個(gè) listener赊窥，會(huì)在流中寫入的所有數(shù)據(jù)刷新到底層資源中時(shí)觸發(fā)。

import {createServer} from 'http'
import Chance from 'chance'

const chance = new Chance()
const server = createServer((req, res) => {
  res.writeHead(200, {'Content-Type': 'text/plain'})
  while (chance.bool({likelihood: 95})) {
    res.write(`${chance.string()}\n`)
  }
  res.end('\n\n')
  res.on('finish', () => console.log('All data sent'))
})

server.listen(8080, () => {
  console.log('listening on http://localhost:8080')
})

上面代碼中 HTTP 服務(wù)里的 res 對(duì)象是一個(gè) http.ServerResponse 對(duì)象糕再，實(shí)際上也是一個(gè) Writable stream拱镐。

實(shí)現(xiàn) Writable stream

import {Writable} from 'stream'
import {promises as fs} from 'fs'

class ToFileStream extends Writable {
  constructor(options) {
    super({...options, objectMode: true})
  }

  _write(chunk, encoding, cb) {
    fs.writeFile(chunk.path, chunk.content)
      .then(() => cb())
      .catch(cb)
  }
}

const tfs = new ToFileStream()

tfs.write({path: 'file1.txt', content: 'Hello'})
tfs.write({path: 'file2.txt', content: 'Node.js'})
tfs.write({path: 'file3.txt', content: 'streams'})
tfs.end(() => console.log('All files created'))

簡化形式

import {Writable} from 'stream'
import {promises as fs} from 'fs'

const tfs = new Writable({
  objectMode: true,
  write(chunk, encoding, cb) {
    fs.writeFile(chunk.path, chunk.content)
      .then(() => cb())
      .catch(cb)
  }
})

tfs.write({path: 'file1.txt', content: 'Hello'})
tfs.write({path: 'file2.txt', content: 'Node.js'})
tfs.write({path: 'file3.txt', content: 'streams'})
tfs.end(() => console.log('All files created'))

Duplex streams

Duplex 流艘款，既 Readable 又 Writable 的流。它的場景在于沃琅，有時(shí)候我們描述的實(shí)體既是數(shù)據(jù)源哗咆，也是數(shù)據(jù)的接收者，比如網(wǎng)絡(luò)套接字益眉。
Duplex 流同時(shí)繼承來著 stream.Readable 和 stream.Writable 的方法晌柬。
為了創(chuàng)建一個(gè)自定義的 Duplex 流，我們必須同時(shí)提供 _read() 和 _write() 的實(shí)現(xiàn)郭脂。

Transform streams

Transform 流是一種特殊類型的 Duplex 流年碘，主要針對(duì)數(shù)據(jù)的轉(zhuǎn)換。
對(duì)于 Duplex 流來說展鸡，流入和流出的數(shù)據(jù)之間并沒有直接的聯(lián)系屿衅。比如一個(gè) TCP 套接字，只是從遠(yuǎn)端接收或者發(fā)送數(shù)據(jù)莹弊，套接字本身不知曉輸入輸出之間的任何關(guān)系涤久。

Duplex stream

而 Transform 流則會(huì)對(duì)收到的每一段數(shù)據(jù)都應(yīng)用某種轉(zhuǎn)換操作，從 Writable 端接收數(shù)據(jù)箱硕，進(jìn)行某種形式地轉(zhuǎn)換后再通過 Readable 端提供給外部拴竹。

Transform stream

實(shí)現(xiàn) Transform 流

import {Transform} from 'stream'

class ReplaceStream extends Transform {
  constructor(searchStr, replaceStr, options) {
    super({...options})
    this.searchStr = searchStr
    this.replaceStr = replaceStr
    this.tail = ''
  }

  _transform(chunk, encoding, callback) {
    const pieces = (this.tail + chunk).split(this.searchStr)
    const lastPiece = pieces[pieces.length - 1]
    const tailLen = this.searchStr.length - 1
    this.tail = lastPiece.slice(-tailLen)
    pieces[pieces.length - 1] = lastPiece.slice(0, -tailLen)
    this.push(pieces.join(this.replaceStr))
    callback()
  }

  _flush(callback) {
    this.push(this.tail)
    callback()
  }
}


const replaceStream = new ReplaceStream('World', 'Node.js')
replaceStream.on('data', chunk => console.log(chunk.toString()))
replaceStream.write('Hello W')
replaceStream.write('orld')
replaceStream.end()

其中核心的 _transform() 方法，其有著和 Writable 流的 _write() 方法基本一致的簽名剧罩，但并不會(huì)將處理后的數(shù)據(jù)寫入底層資源，而是通過 this.push() 推送給內(nèi)部 buffer座泳，正如 Readable 流中 _read() 方法的行為惠昔。
所以形成了 Transform 流整體上接收幕与、轉(zhuǎn)換、發(fā)送的行為镇防。
_flush() 則會(huì)在流結(jié)束前調(diào)用啦鸣。

簡化形式

import {Transform} from 'stream'

const searchStr = 'World'
const replaceStr = 'Node.js'
let tail = ''

const replaceStream = new Transform({
  defaultEncoding: 'utf-8',

  transform(chunk, encoding, cb) {
    const pieces = (tail + chunk).split(searchStr)
    const lastPiece = pieces[pieces.length - 1]
    const tailLen = searchStr.length - 1
    tail = lastPiece.slice(-tailLen)
    pieces[pieces.length - 1] = lastPiece.slice(0, -tailLen)
    this.push(pieces.join(replaceStr))
    cb()
  },
  flush(cb) {
    this.push(tail)
    cb()
  }
})
replaceStream.on('data', chunk => console.log(chunk.toString()))
replaceStream.write('Hello W')
replaceStream.write('orld')
replaceStream.end()

Transform 流篩選和聚合數(shù)據(jù)

數(shù)據(jù)源 data.csv：

type,country,profit
Household,Namibia,597290.92
Baby Food,Iceland,808579.10
Meat,Russia,277305.60
Meat,Italy,413270.00
Cereal,Malta,174965.25
Meat,Indonesia,145402.40
Household,Italy,728880.54

package.json:

{
  "type": "module",
  "main": "index.js",
  "dependencies": {
    "csv-parse": "^4.10.1"
  },
  "engines": {
    "node": ">=14"
  },
  "engineStrict": true
}

FilterByCountry Transform 流 filter-by-country.js：

import {Transform} from 'stream'

export class FilterByCountry extends Transform {
  constructor(country, options = {}) {
    options.objectMode = true
    super(options)
    this.country = country
  }

  _transform(record, enc, cb) {
    if (record.country === this.country) {
      this.push(record)
    }
    cb()
  }
}

SumProfit Transform 流 sum-profit.js：

import {Transform} from 'stream'

export class SumProfit extends Transform {
  constructor(options = {}) {
    options.objectMode = true
    super(options)
    this.total = 0
  }

  _transform(record, enc, cb) {
    this.total += Number.parseFloat(record.profit)
    cb()
  }

  _flush(cb) {
    this.push(this.total.toString())
    cb()
  }
}

index.js：

import {createReadStream} from 'fs'
import parse from 'csv-parse'
import {FilterByCountry} from './filter-by-conutry.js'
import {SumProfit} from './sum-profit.js'

const csvParser = parse({columns: true})

createReadStream('data.csv')
  .pipe(csvParser)
  .pipe(new FilterByCountry('Italy'))
  .pipe(new SumProfit())
  .pipe(process.stdout)

參考資料

Node.js Design Patterns: Design and implement production-grade Node.js applications using proven patterns and techniques, 3rd Edition