node中的流（stream）

時間 2020-05-30

標籤 node stream 欄目 Apache 简体版

原文原文鏈接

官方的定義html

流（stream）是 Node.js 中處理流式數據的抽象接口。stream模塊用於構建實現了流接口的對象。node

咱們通常是直接使用node提供的流對象，例如在服務器請求、文件模塊中使用。api

流的分類

Writable- 可寫入數據的流（例如fs.createWriteStream()）。
Readable- 可讀取數據的流（例如fs.createReadStream()）。
Duplex- 可讀又可寫的流（例如net.Socket）。
Transform- 在讀寫過程當中能夠修改或轉換數據的Duplex流（例如zlib.createDeflate()）

流的重要事件和方法

下面會結合具體的例子來梳理一下流的經常使用事件和方法，加深對流的理解。bash

一、data和end事件

流類型：可讀流 data事件在可讀流將數據傳給消費者後觸發，特別注意的是，添加了該事件的流會自動切換爲流動模式， end事件在當流中沒有數據可供消費時觸發:服務器

const stream = fs.createReadStream('./file.txt') // 默認靜止態
let chunks = []
stream.on("data", (chunk) => { // 變成流動態
    chunks.push(chunk)
})
stream.on("end", () => {
    constcontent = Buffer.concat(chunks).toString()
    console.log(content)
})
複製代碼

其中chunk是buffer類型 補充，在任意時刻，可讀流會處於如下三種狀態之一：學習

readable.readableFlowing === null
readable.readableFlowing === false
readable.readableFlowing === true 初始時則readable.readableFlowing爲null，添加data事件後變爲true。調用readable.pause()、readable.unpipe()、或接收到背壓，則readable.readableFlowing會被設爲false，在這個狀態下，爲data事件綁定監聽器不會使readable.readableFlowing切換到 true

二、readable事件與read()方法

流類型：可讀流 readable事件代表流有新的動態：要麼有新的數據，要麼到達流的盡頭。下面是讀取文件的例子：測試

const stream = fs.createReadStream('./file.txt')
let chunks = []
// stream中無數據也會觸發readable， 此時read方法獲得null.
// 讀取到流的盡頭也會觸發，而且在end事件以前
stream.on("readable", () => {
    console.log('觸發readable');
    let data;
    while (data = stream.read(1024)) {
        chunks.push(data)
        console.log('讀取數據', data);
    }
})
stream.on("end", () => {
    const content = Buffer.concat(chunks).toString();
    console.log(content)
})
複製代碼

**使用readable會使流的狀態變成暫停模式，即便監聽了data事件。在調用read方法且有返回數據時會觸發data事件。**上面代碼中，read方法讀取內部緩衝中的數據，若是不指定size參數，則是讀取內部緩衝中的全部數據，注意不是流中的全部數據，不指定size的話也就不必使用while循環了，直接一次性讀取，while循環代碼塊可變爲：優化

data = stream.read()
data && chunks.push(data)
console.log('讀取數據', data);
複製代碼

根據運行結果，read方法將緩衝區數據讀完後會觸發readable事件，也就是當read()返回null後觸發。這是第二種讀取可讀流的模式，即經過read()讀取ui

三、pipe()和unpipe()

流類型：可讀流定義見官方文檔，下面例子使用pipe響應http請求spa

const http = require('http')
const fs = require('fs')
const server = http.createServer()
server.on('request', (request, response) => {
    const stream = fs.createReadStream('./file.txt')
    stream.pipe(response)
})
server.listen(8888)
複製代碼

使用pipe時數據流會被自動管理，因此即便可讀流更快，目標可寫流也不會超負荷。 另外pipe()會返回目標流的引用，支持鏈式操做，假設b是個轉換流： a.pipe(b).pipe(c)` unpipe()則是用於解綁以前綁定的可寫流。上面的例子能夠用data事件改寫：

//....
server.on('request', (request, response) => {
    const stream = fs.createReadStream('./file.txt')
    stream.on("data", (chunk) => {
        response.write(chunk)
    })
    stream.on("end", () => {
        response.end() // 使用pipe的話默認會在可讀流觸發end事件後調用end()結束寫入
    })
})
//...
複製代碼

不過這樣寫可能會讓可寫流超負荷，這就要引入drain的概念了

四、drain和finish事件

流類型：可寫流若是可寫流調用write() 返回 false，說明寫的太快了，不能再往裏面寫了。當能夠繼續寫入數據到流時會觸發 'drain' 事件。對於上面的例子，咱們來測試一下drain事件是否觸發：

//....
server.on('request', (request, response) => {
    const stream = fs.createReadStream('./file.txt')
    stream.on("data", (chunk) => {
        response.write(chunk)
    })
    stream.on("end", () => {
        response.end()
    })
    response.on("drain", () => {
        console.log('能夠寫了')
  })
})
//...
複製代碼

文件file.txt大小100kb左右，運行後看到drain事件觸發了4次。雖然寫的太快了，可是從http響應的結果看，數據並無丟失。查了一下文檔看到有這樣的說明：

當流還未被排空時，調用write()會緩衝chunk，並返回false。一旦全部當前緩衝的數據塊都被排空了（被操做系統接收並傳輸），則觸發'drain'事件。建議一旦write()返回 false，則再也不寫入任何數據塊，直到'drain'事件被觸發。當流還未被排空時，也是能夠調用write()，Node.js 會緩衝全部被寫入的數據塊，直到達到最大內存佔用，這時它會無條件停止。

因此write()返回 false時就不要再往裏面寫數據了。上面的例子能夠這樣優化：

server.on('request', (request, response) => {
    conststream = fs.createReadStream('./file.txt')
    let ok = true
    stream.on("data", (chunk) => {
        ok = response.write(chunk)
        if(!ok) {
            stream.pause()
            console.log('別寫了')
            ok = true
        }
    })
    stream.on("end", () => {
        response.end()
    })
    response.on("drain", () => {
        console.log('能夠寫了')
        stream.resume()
    })
})
複製代碼