簡單理解 backpressure（背壓）機制

前言

本文僅是做者本身的看法，若有錯誤還請即便指正

爲何存在背壓機制？

咱們首先來看一段代碼, 這段代碼存在什麼問題？

乍一看，感受沒啥大毛病，可是若是writable.write()寫入數據比較慢，可是可讀流又在不斷的傳輸數據，就會形成內存溢出，造成阻塞。

const fs = require('fs')

const readable = fs.createReadStream('./小婦人.mp4')
const writable = fs.createWriteStream('./小婦人(1).mp4')

readable.on('data', (chunk) => {
    // 這裏存在問題↓↓↓↓↓↓↓
    writable.write(chunk);
})

readable.on('end', () => {
    writable.end()
})
複製代碼

流的錯誤處理

若是可寫流，沒法正確的處理大量由可讀流傳輸的數據，可讀流並不會被銷燬，這會致使咱們寫入的文件被損壞。咱們必須添加適當的錯誤處理程序，在當流發生故障的時候，銷燬管道中的全部流。

const gzip = require('zlib').createGzip();
const fs = require('fs');

const readable = fs.createReadStream('好萊塢往事.1080p.mkv');
const writable = fs.createWriteStream('好萊塢往事.1080p.mkv.gz');

// 若是可寫流發生故障，壓縮文件會壓縮失敗
readable.pipe(gzip).pipe(writable);
複製代碼

在 Node 8.x 版本以前咱們使用 pump。對於更高版本的 Node, 可使用pipeline。

const gzip = require('zlib').createGzip();
const { pipeline } = require('stream')
const fs = require('fs');

const readable = fs.createReadStream('好萊塢往事.1080p.mkv');
const writable = fs.createWriteStream('好萊塢往事.1080p.mkv.gz');

pipeline(
    readable,
    gzip,
    writable,
    error => {
        if (error) {
            console.log('電影壓縮失敗')
        } else {
            console.log('電影壓縮成功')
        }
    }
)
複製代碼

咱們也可使用 promisify 將其改形成 async/await 的形式。

const gzip = require('zlib').createGzip()
const { pipeline } = require('stream')
const { promisify } = require('util')
const fs = require('fs')
const readable = fs.createReadStream('好萊塢往事.1080p.mkv')
const writable = fs.createWriteStream('好萊塢往事.1080p.mkv.gz')
const asyncPipeline = promisify(pipeline)

async function start () {
    try {
        await asyncPipeline(
            readable,
            gzip,
            writable
        )
        console.log('電影壓縮成功')
    } catch (error) {
        console.log('電影壓縮失敗')
    }
}

start()
複製代碼

可讀流太快了

硬盤的寫入速度，遠遠小於硬盤的讀取速度。若是可讀流太快，而可寫流的沒法迅速的消費可讀流傳輸的數據，寫入流將會把 chunk，push 到寫隊列中方便以後使用，這樣就會形成數據在內存中的累積。這個時候將會觸發 backpressur（背壓） 機制。若是沒有 backpressur（背壓） 機制，系統將會出現以下的問題：

1. 內存溢出
2. 其餘進程變得緩慢
3. 垃圾收集器將超負荷運做

背壓機制是如何解決這些問題的？

在代碼中調用pipe時，它會向可寫流發出信號，表示有數據準備傳輸。當咱們的可寫流使用 write() 寫入數據時，若是寫隊列繁忙，或者內部緩存區已經溢出了，write() 將會返回false。

這個時候，背壓機制就會啓動，它會暫停任何數據傳入到可寫流中，並等待可寫流準備好，清空內部緩存區後。將會發出 drain 事件，並恢復可讀流的傳輸。

這就意味着 pipe 只會使用固定大小的內存，不會存在內存泄漏的問題。

爲何咱們平時不多關注背壓的問題呢？那是由於在你調用 pipe 時，Node.js已經自動處理了這些問題。可是若是咱們須要實現自定義流，則須要考慮到這些問題。

參考

• Backpressuring in Streams