日本伦理视频网,日本三级人妻三级

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)勢(shì)：組合性。

空間效率

使用 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ì)來自于其接口和可組合性，允許我們實(shí)現(xiàn)干凈、優(yōu)雅、簡(jiǎn)潔的代碼。對(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è)長(zhǎng)度為 size 的隨機(jī)字符串
通過 push() 方法將字符傳推送到內(nèi)部 buffer
依據(jù) 5% 的幾率自行終止，終止時(shí)推送 null 到內(nèi)部 buffer，作為 stream 的結(jié)束標(biāo)志

簡(jiǎn)化版實(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'))

簡(jiǎn)化形式

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 的流。它的場(chǎng)景在于，有時(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)用。

簡(jiǎn)化形式

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

色偷偷精品伊人,欧洲久久精品,欧美综合婷婷骚逼,国产AV主播,国产最新探花在线,九色在线视频一区,伊人大交九欧美,1769亚洲,黄色成人av

Node.js 設(shè)計(jì)模式筆記 —— Streams 流編程

Node.js 設(shè)計(jì)模式筆記 —— Streams 流編程

Buffering vs streaming

空間效率

時(shí)間效率

組合性

客戶端加密

服務(wù)端加密

Streams 詳解

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

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

non-flowing mode

flowing mode

Async iterators

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

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

Writable streams

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

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

Duplex streams

Transform streams

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

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

參考資料

相關(guān)閱讀更多精彩內(nèi)容

友情鏈接更多精彩內(nèi)容

色偷偷精品伊人,欧洲久久精品,欧美综合婷婷骚逼,国产AV主播,国产最新探花在线,九色在线视频一区,伊人大交九 欧美,1769亚洲,黄色成人av

Node.js 設(shè)計(jì)模式筆記 —— Streams 流編程

Buffering vs streaming

空間效率

時(shí)間效率

組合性

客戶端加密

服務(wù)端加密

Streams 詳解

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

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

non-flowing mode

flowing mode

Async iterators

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

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

Writable streams

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

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

Duplex streams

Transform streams

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

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

參考資料

相關(guān)閱讀更多精彩內(nèi)容

友情鏈接更多精彩內(nèi)容

色偷偷精品伊人,欧洲久久精品,欧美综合婷婷骚逼,国产AV主播,国产最新探花在线,九色在线视频一区,伊人大交九欧美,1769亚洲,黄色成人av