Node.js v26.0.0 說明文件

串流 (Stream)#

串流 (stream) 是 Node.js 中處理串流資料的抽象介面。node:stream 模組提供了實作此串流介面的 API。

Node.js 提供了許多串流物件。例如，對 HTTP 伺服器的請求與 process.stdout 都是串流實例。

串流可以是可讀、可寫，或兩者兼具。所有串流都是 EventEmitter 的實例。

存取 node:stream 模組

const stream = require('node:stream');
copy

node:stream 模組對於建立新類型的串流實例非常有用。通常不需要使用 node:stream 模組來消耗 (consume) 串流。

本文件架構#

本文件包含兩個主要章節與第三個補充說明章節。第一部分說明如何在應用程式中使用現有的串流。第二部分說明如何建立新類型的串流。

串流類型#

Node.js 中有四種基本的串流類型：

• Writable：可寫入資料的串流（例如 fs.createWriteStream()）。
• Readable：可從中讀取資料的串流（例如 fs.createReadStream()）。
• Duplex：同時為 Readable 和 Writable 的串流（例如 net.Socket）。
• Transform：在寫入和讀取時可以修改或轉換資料的 Duplex 串流（例如 zlib.createDeflate()）。

此外，此模組還包含公用函式 stream.duplexPair()、stream.pipeline()、stream.finished()、stream.Readable.from() 與 stream.addAbortSignal()。

串流 Promise API#

stream/promises API 為串流提供了一組替代的非同步公用函式，它們會回傳 Promise 物件而非使用回呼 (callback)。此 API 可透過 require('node:stream/promises') 或 require('node:stream').promises 存取。

stream.pipeline(streams[, options])#

stream.pipeline(source[, ...transforms], destination[, options])#

• streams <Stream[]> | <Iterable[]> | <AsyncIterable[]> | <Function[]>
• source <Stream> | <Iterable> | <AsyncIterable> | <Function>
  • 回傳：<Promise> | <AsyncIterable>
• ...transforms <Stream> | <Function>
  • source <AsyncIterable>
  • 回傳：<Promise> | <AsyncIterable>
• destination <Stream> | <Function>
  • source <AsyncIterable>
  • 回傳：<Promise> | <AsyncIterable>
• options <Object> 管道選項
  • signal <AbortSignal>
  • end <boolean> 當來源串流結束時，是否結束目標串流。轉換串流 (Transform streams) 總是會結束，即使此值為 false。預設值： true。
• 回傳：<Promise> 在管道完成時履行 (fulfill)。

const { pipeline } = require('node:stream/promises');
const fs = require('node:fs');
const zlib = require('node:zlib');

async function run() {
  await pipeline(
    fs.createReadStream('archive.tar'),
    zlib.createGzip(),
    fs.createWriteStream('archive.tar.gz'),
  );
  console.log('Pipeline succeeded.');
}

run().catch(console.error);
import { pipeline } from 'node:stream/promises';
import { createReadStream, createWriteStream } from 'node:fs';
import { createGzip } from 'node:zlib';

await pipeline(
  createReadStream('archive.tar'),
  createGzip(),
  createWriteStream('archive.tar.gz'),
);
console.log('Pipeline succeeded.');
copy

若要使用 AbortSignal，請將其放入選項物件中作為最後一個參數傳遞。當信號被中止時，底層管道將被呼叫 destroy，並帶有 AbortError。

const { pipeline } = require('node:stream/promises');
const fs = require('node:fs');
const zlib = require('node:zlib');

async function run() {
  const ac = new AbortController();
  const signal = ac.signal;

  setImmediate(() => ac.abort());
  await pipeline(
    fs.createReadStream('archive.tar'),
    zlib.createGzip(),
    fs.createWriteStream('archive.tar.gz'),
    { signal },
  );
}

run().catch(console.error); // AbortError
import { pipeline } from 'node:stream/promises';
import { createReadStream, createWriteStream } from 'node:fs';
import { createGzip } from 'node:zlib';

const ac = new AbortController();
const { signal } = ac;
setImmediate(() => ac.abort());
try {
  await pipeline(
    createReadStream('archive.tar'),
    createGzip(),
    createWriteStream('archive.tar.gz'),
    { signal },
  );
} catch (err) {
  console.error(err); // AbortError
}
copy

pipeline API 也支援非同步產生器：

const { pipeline } = require('node:stream/promises');
const fs = require('node:fs');

async function run() {
  await pipeline(
    fs.createReadStream('lowercase.txt'),
    async function* (source, { signal }) {
      source.setEncoding('utf8');  // Work with strings rather than `Buffer`s.
      for await (const chunk of source) {
        yield await processChunk(chunk, { signal });
      }
    },
    fs.createWriteStream('uppercase.txt'),
  );
  console.log('Pipeline succeeded.');
}

run().catch(console.error);
import { pipeline } from 'node:stream/promises';
import { createReadStream, createWriteStream } from 'node:fs';

await pipeline(
  createReadStream('lowercase.txt'),
  async function* (source, { signal }) {
    source.setEncoding('utf8');  // Work with strings rather than `Buffer`s.
    for await (const chunk of source) {
      yield await processChunk(chunk, { signal });
    }
  },
  createWriteStream('uppercase.txt'),
);
console.log('Pipeline succeeded.');
copy

請記得處理傳入非同步產生器的 signal 參數。特別是在非同步產生器作為管道來源（即第一個參數）的情況下，否則管道可能永遠無法完成。

const { pipeline } = require('node:stream/promises');
const fs = require('node:fs');

async function run() {
  await pipeline(
    async function* ({ signal }) {
      await someLongRunningfn({ signal });
      yield 'asd';
    },
    fs.createWriteStream('uppercase.txt'),
  );
  console.log('Pipeline succeeded.');
}

run().catch(console.error);
import { pipeline } from 'node:stream/promises';
import fs from 'node:fs';
await pipeline(
  async function* ({ signal }) {
    await someLongRunningfn({ signal });
    yield 'asd';
  },
  fs.createWriteStream('uppercase.txt'),
);
console.log('Pipeline succeeded.');
copy

pipeline API 亦提供 回呼版本。

stream.finished(stream[, options])#

• stream <Stream> | <ReadableStream> | <WritableStream> 一個可讀及/或可寫的串流/網頁串流 (webstream)。
• options <Object>
  • error <boolean> | <undefined>
  • readable <boolean> | <undefined>
  • writable <boolean> | <undefined>
  • signal <AbortSignal> | <undefined>
  • cleanup <boolean> | <undefined> 若為 true，則在 promise 履行前移除由此函式註冊的接聽器。預設值： false。
• 回傳：<Promise> 當串流不再可讀或可寫時履行。

const { finished } = require('node:stream/promises');
const fs = require('node:fs');

const rs = fs.createReadStream('archive.tar');

async function run() {
  await finished(rs);
  console.log('Stream is done reading.');
}

run().catch(console.error);
rs.resume(); // Drain the stream.
import { finished } from 'node:stream/promises';
import { createReadStream } from 'node:fs';

const rs = createReadStream('archive.tar');

async function run() {
  await finished(rs);
  console.log('Stream is done reading.');
}

run().catch(console.error);
rs.resume(); // Drain the stream.
copy

finished API 亦提供 回呼版本。

在回傳的 promise 被解析 (resolved) 或拒絕 (rejected) 後，stream.finished() 會留下懸掛的事件接聽器（特別是 'error'、'end'、'finish' 與 'close'）。這樣做是為了避免非預期的 'error' 事件（由於錯誤的串流實作）導致程式異常崩潰。如果不希望有此行為，應將 options.cleanup 設定為 true。

await finished(rs, { cleanup: true });
copy

物件模式#

Node.js API 建立的所有串流僅操作字串、<Buffer>、<TypedArray> 與 <DataView> 物件。

• Strings 與 Buffers 是串流中最常用的類型。
• TypedArray 與 DataView 讓您可以使用 Int32Array 或 Uint8Array 等類型處理二進位資料。當您將 TypedArray 或 DataView 寫入串流時，Node.js 會處理原始位元組。

然而，串流實作可以處理其他類型的 JavaScript 值（null 除外，它在串流中具有特殊用途）。此類串流被認為運作於「物件模式」。

在建立串流時使用 objectMode 選項可將串流實例切換到物件模式。嘗試將現有串流切換到物件模式是不安全的。

緩衝#

Writable 與 Readable 串流都會將資料儲存在內部緩衝區中。

潛在緩衝的資料量取決於傳入串流建構函式的 highWaterMark 選項。對於一般串流，highWaterMark 選項指定了位元組總數。對於在物件模式下運作的串流，highWaterMark 指定了物件總數。對於操作字串（但未解碼）的串流，highWaterMark 指定了 UTF-16 編碼單元的總數。

當實作呼叫 stream.push(chunk) 時，資料會緩存在 Readable 串流中。如果串流的使用者未呼叫 stream.read()，資料將留在內部隊列中直到被消耗。

一旦內部讀取緩衝區的總大小達到 highWaterMark 指定的閾值，串流將暫時停止從底層資源讀取資料，直到目前緩衝的資料可以被消耗（也就是說，串流將停止呼叫用於填充讀取緩衝區的內部 readable._read() 方法）。

當重複呼叫 writable.write(chunk) 方法時，資料會緩存在 Writable 串流中。當內部寫入緩衝區的總大小低於 highWaterMark 設定的閾值時，呼叫 writable.write() 將回傳 true。一旦內部緩衝區的大小達到或超過 highWaterMark，將回傳 false。

stream API 的一個關鍵目標，特別是 stream.pipe() 方法，是將資料緩衝限制在可接受的水平，使得速度不同的來源和目標不會耗盡可用記憶體。

highWaterMark 選項是一個閾值，而不是限制：它規定了串流在停止請求更多資料之前緩衝的資料量。它通常不強制執行嚴格的記憶體限制。特定的串流實作可以選擇強制執行更嚴格的限制，但這是選配的。

由於 Duplex 與 Transform 串流同時是 Readable 與 Writable，因此各別維護了*兩個*獨立的內部緩衝區用於讀取和寫入，允許各側獨立運作，同時保持適當且高效的資料流。例如，net.Socket 實例是 Duplex 串流，其 Readable 側允許消耗*從* socket 接收的資料，而其 Writable 側允許將資料寫入*至* socket。由於寫入 socket 的速度可能比接收資料的速度快或慢，因此各側應獨立運作（和緩衝）。

內部緩衝機制是內部實作細節，隨時可能更改。然而，對於某些進階實作，可以使用 writable.writableBuffer 或 readable.readableBuffer 取得內部緩衝區。不鼓勵使用這些未公開的屬性。

給串流使用者的 API#

幾乎所有的 Node.js 應用程式，無論多麼簡單，都會以某種方式使用串流。以下是在實作 HTTP 伺服器的 Node.js 應用程式中使用串流的範例：

const http = require('node:http');

const server = http.createServer((req, res) => {
  // `req` is an http.IncomingMessage, which is a readable stream.
  // `res` is an http.ServerResponse, which is a writable stream.

  let body = '';
  // Get the data as utf8 strings.
  // If an encoding is not set, Buffer objects will be received.
  req.setEncoding('utf8');

  // Readable streams emit 'data' events once a listener is added.
  req.on('data', (chunk) => {
    body += chunk;
  });

  // The 'end' event indicates that the entire body has been received.
  req.on('end', () => {
    try {
      const data = JSON.parse(body);
      // Write back something interesting to the user:
      res.write(typeof data);
      res.end();
    } catch (er) {
      // uh oh! bad json!
      res.statusCode = 400;
      return res.end(`error: ${er.message}`);
    }
  });
});

server.listen(1337);

// $ curl localhost:1337 -d "{}"
// object
// $ curl localhost:1337 -d "\"foo\""
// string
// $ curl localhost:1337 -d "not json"
// error: Unexpected token 'o', "not json" is not valid JSON
copy

Writable 串流（如範例中的 res）公開了 write() 與 end() 等方法，用於將資料寫入串流。

Readable 串流使用 EventEmitter API 來通知應用程式程式碼何時可以從串流中讀取資料。這些可用資料可以透過多種方式從串流中讀取。

Writable 與 Readable 串流都以各種方式使用 EventEmitter API 來傳達串流的目前狀態。

Duplex 與 Transform 串流同時是 Writable 與 Readable。

正在將資料寫入串流或從串流消耗資料的應用程式不需要直接實作串流介面，通常也沒有理由呼叫 require('node:stream')。

希望實作新類型串流的開發人員應參考 給串流實作者的 API 章節。

可寫串流 (Writable streams)#

可寫串流是資料寫入之*目的地*的抽象化。

Writable 串流的例子包括：

• 用戶端上的 HTTP 請求
• 伺服器上的 HTTP 回應
• fs 寫入串流
• zlib 串流
• crypto 串流
• TCP sockets
• 子程序 stdin
• process.stdout, process.stderr

其中一些範例實際上是實作了 Writable 介面的 Duplex 串流。

所有 Writable 串流都實作了由 stream.Writable 類別定義的介面。

雖然特定的 Writable 串流實例在各方面可能有所不同，但所有 Writable 串流都遵循相同的基本用法模式，如下例所示：

const myStream = getWritableStreamSomehow();
myStream.write('some data');
myStream.write('some more data');
myStream.end('done writing data');
copy

類別：stream.Writable#

事件：'close'#

當串流及其任何底層資源（例如檔案描述符）已關閉時，會發出 'close' 事件。該事件表示將不再發出任何事件，也不會再進行任何計算。

如果 Writable 串流是在帶有 emitClose 選項的情況下建立的，則它始終會發出 'close' 事件。

事件：'drain'#

如果對 stream.write(chunk) 的呼叫回傳 false，則當適合恢復向串流寫入資料時，將發出 'drain' 事件。

// Write the data to the supplied writable stream one million times.
// Be attentive to back-pressure.
function writeOneMillionTimes(writer, data, encoding, callback) {
  let i = 1000000;
  write();
  function write() {
    let ok = true;
    do {
      i--;
      if (i === 0) {
        // Last time!
        writer.write(data, encoding, callback);
      } else {
        // See if we should continue, or wait.
        // Don't pass the callback, because we're not done yet.
        ok = writer.write(data, encoding);
      }
    } while (i > 0 && ok);
    if (i > 0) {
      // Had to stop early!
      // Write some more once it drains.
      writer.once('drain', write);
    }
  }
}
copy

事件：'error'#

• 類型：<Error>

如果在寫入或導流資料時發生錯誤，則會發出 'error' 事件。呼叫時會向接聽器回呼傳遞單個 Error 參數。

除非在建立串流時將 autoDestroy 選項設定為 false，否則發出 'error' 事件時串流會關閉。

在 'error' 之後，除了 'close' 之外*不應*再發出任何其他事件（包括 'error' 事件）。

事件：'finish'#

在呼叫 stream.end() 方法且所有資料都已排空 (flushed) 到底層系統後，會發出 'finish' 事件。

const writer = getWritableStreamSomehow();
for (let i = 0; i < 100; i++) {
  writer.write(`hello, #${i}!\n`);
}
writer.on('finish', () => {
  console.log('All writes are now complete.');
});
writer.end('This is the end\n');
copy

事件：'pipe'#

• src <stream.Readable> 正在導流到此可寫串流的來源串流

當在可讀串流上呼叫 stream.pipe() 方法並將此可寫串流新增到其目標集合時，會發出 'pipe' 事件。

const writer = getWritableStreamSomehow();
const reader = getReadableStreamSomehow();
writer.on('pipe', (src) => {
  console.log('Something is piping into the writer.');
  assert.equal(src, reader);
});
reader.pipe(writer);
copy

事件：'unpipe'#

• src <stream.Readable> 對此可寫串流執行了 unpipe 的來源串流

當在 Readable 串流上呼叫 stream.unpipe() 方法，將此 Writable 從其目標集合中移除時，會發出 'unpipe' 事件。

當 Readable 串流導流到此 Writable 串流但後者發出錯誤時，也會發出此事件。

const writer = getWritableStreamSomehow();
const reader = getReadableStreamSomehow();
writer.on('unpipe', (src) => {
  console.log('Something has stopped piping into the writer.');
  assert.equal(src, reader);
});
reader.pipe(writer);
reader.unpipe(writer);
copy

writable.cork()#

writable.cork() 方法強制將所有寫入的資料緩存在記憶體中。當呼叫 stream.uncork() 或 stream.end() 方法時，緩存的資料將被排空 (flushed)。

writable.cork() 的主要目的是應對在極短時間內連續向串流寫入多個小區塊的情況。writable.cork() 不會立即將它們轉發到底層目的地，而是緩存所有區塊，直到呼叫 writable.uncork()，這會將它們全部傳遞給 writable._writev() (如果存在)。這可以防止在等待處理第一個小區塊時發生隊頭阻塞 (head-of-line blocking)。但是，在未實作 writable._writev() 的情況下使用 writable.cork() 可能會對吞吐量產生不利影響。

另請參閱：writable.uncork(), writable._writev()。

writable.destroy([error])#

• error <Error> 選配，隨 'error' 事件發出的錯誤。
• 傳回：<this>

銷毀串流。選配地發出 'error' 事件，並發出 'close' 事件（除非 emitClose 設定為 false）。在此呼叫之後，可寫串流已結束，隨後對 write() 或 end() 的呼叫將導致 ERR_STREAM_DESTROYED 錯誤。這是一種破壞性且立即銷毀串流的方法。之前的 write() 呼叫可能尚未排空，且可能觸發 ERR_STREAM_DESTROYED 錯誤。如果資料應在關閉前排空，請使用 end() 而非 destroy，或者在銷毀串流前等待 'drain' 事件。

const { Writable } = require('node:stream');

const myStream = new Writable();

const fooErr = new Error('foo error');
myStream.destroy(fooErr);
myStream.on('error', (fooErr) => console.error(fooErr.message)); // foo error
copy

const { Writable } = require('node:stream');

const myStream = new Writable();

myStream.destroy();
myStream.on('error', function wontHappen() {});
copy

const { Writable } = require('node:stream');

const myStream = new Writable();
myStream.destroy();

myStream.write('foo', (error) => console.error(error.code));
// ERR_STREAM_DESTROYED
copy

一旦呼叫了 destroy()，任何進一步的呼叫都將是空操作，除了來自 _destroy() 的錯誤外，不會再發出進一步的 'error'。

實作編寫者不應覆寫此方法，而應實作 writable._destroy()。

writable.closed#

• 類型：<boolean>

在 'close' 發出後為 true。

writable.destroyed#

• 類型：<boolean>

在呼叫 writable.destroy() 後為 true。

const { Writable } = require('node:stream');

const myStream = new Writable();

console.log(myStream.destroyed); // false
myStream.destroy();
console.log(myStream.destroyed); // true
copy

writable.end([chunk[, encoding]][, callback])#

• chunk <string> | <Buffer> | <TypedArray> | <DataView> | <any> 選配的寫入資料。對於不在物件模式下運作的串流，chunk 必須是 <string>, <Buffer>, <TypedArray> 或 <DataView>。對於物件模式串流，chunk 可以是除 null 以外的任何 JavaScript 值。
• encoding <string> 如果 chunk 是字串，則為其編碼格式
• callback <Function> 串流結束時的回呼。
• 傳回：<this>

呼叫 writable.end() 方法表示不會再向 Writable 寫入更多資料。選配的 chunk 與 encoding 參數允許在關閉串流前立即寫入最後一塊資料。

在呼叫 stream.end() 之後呼叫 stream.write() 方法將會引發錯誤。

// Write 'hello, ' and then end with 'world!'.
const fs = require('node:fs');
const file = fs.createWriteStream('example.txt');
file.write('hello, ');
file.end('world!');
// Writing more now is not allowed!
copy

writable.setDefaultEncoding(encoding)#

• encoding <string> 新的預設編碼
• 傳回：<this>

writable.setDefaultEncoding() 方法為 Writable 串流設定預設 encoding。

writable.uncork()#

writable.uncork() 方法會排空 (flush) 自呼叫 stream.cork() 以來緩存的所有資料。

當使用 writable.cork() 與 writable.uncork() 來管理串流寫入的緩衝時，請使用 process.nextTick() 延遲呼叫 writable.uncork()。這樣做可以批次處理在給定 Node.js 事件迴圈階段中發生的所有 writable.write() 呼叫。

stream.cork();
stream.write('some ');
stream.write('data ');
process.nextTick(() => stream.uncork());
copy

如果在串流上多次呼叫 writable.cork() 方法，則必須呼叫相同次數的 writable.uncork() 才能完全排空緩存的資料。

stream.cork();
stream.write('some ');
stream.cork();
stream.write('data ');
process.nextTick(() => {
  stream.uncork();
  // The data will not be flushed until uncork() is called a second time.
  stream.uncork();
});
copy

另請參閱：writable.cork()。

writable.writable#

• 類型：<boolean>

如果呼叫 writable.write() 是安全的（表示串流尚未被銷毀、報錯或結束），則為 true。

writable.writableAborted#

• 類型：<boolean>

回傳串流是否在發出 'finish' 之前已被銷毀或報錯。

writable.writableEnded#

• 類型：<boolean>

在呼叫 writable.end() 後為 true。此屬性不表示資料是否已排空，若要判斷此項，請改用 writable.writableFinished。

writable.writableCorked#

• 類型：<integer>

需要呼叫 writable.uncork() 的次數，以便完全取消串流的緩衝 (uncork)。

writable.errored#

• 類型：<Error>

如果串流在銷毀時帶有錯誤，則回傳該錯誤。

writable.writableFinished#

• 類型：<boolean>

在發出 'finish' 事件前立即設定為 true。

writable.writableHighWaterMark#

• 類型：<number>

回傳建立此 Writable 時傳入的 highWaterMark 值。

writable.writableLength#

• 類型：<number>

此屬性包含隊列中準備寫入的位元組數（或物件數）。此值提供了關於 highWaterMark 狀態的內省資料。

writable.writableNeedDrain#

• 類型：<boolean>

如果串流的緩衝區已滿且串流將發出 'drain'，則為 true。

writable.writableObjectMode#

• 類型：<boolean>

給定 Writable 串流之 objectMode 屬性的 getter。

writable[Symbol.asyncDispose]()#

以 AbortError 呼叫 writable.destroy() 並回傳一個在串流結束時履行的 promise。

writable.write(chunk[, encoding][, callback])#

• chunk <string> | <Buffer> | <TypedArray> | <DataView> | <any> 選配的寫入資料。對於不在物件模式下運作的串流，chunk 必須是 <string>, <Buffer>, <TypedArray> 或 <DataView>。對於物件模式串流，chunk 可以是除 null 以外的任何 JavaScript 值。
• encoding <string> | <null> 如果 chunk 是字串，則為其編碼。預設值： 'utf8'
• callback <Function> 當此資料區塊排空後的回呼。
• 回傳：<boolean> 如果串流希望呼叫端程式碼在繼續寫入額外資料前等待 'drain' 事件發出，則回傳 false；否則回傳 true。

writable.write() 方法將一些資料寫入串流，並在資料被完全處理後呼叫提供的 callback。如果發生錯誤，callback 將以錯誤作為其第一個參數被呼叫。callback 是非同步呼叫的，且在 'error' 發出之前。

如果在接受 chunk 後，內部緩衝區小於建立串流時配置的 highWaterMark，則回傳值為 true。如果回傳 false，則應停止進一步嘗試向串流寫入資料，直到發出 'drain' 事件。

當串流未處於消耗 (draining) 狀態時，呼叫 write() 會緩存 chunk 並回傳 false。一旦所有目前緩存的區塊被消耗完畢（作業系統接受傳遞），將發出 'drain' 事件。一旦 write() 回傳 false，在發出 'drain' 事件前請勿再寫入更多區塊。雖然允許在未消耗完畢的串流上呼叫 write()，但 Node.js 會緩存所有寫入的區塊，直到達到最大記憶體使用量，屆時它將無條件中止。即使在中止前，高記憶體使用量也會導致垃圾回收器效能不佳以及高 RSS（通常即使在不再需要記憶體後也不會釋放回系統）。由於如果遠端對等端不讀取資料，TCP socket 可能永遠不會消耗完畢，因此寫入未在消耗狀態的 socket 可能會導致遠端可利用的漏洞。

在串流未處於消耗狀態時寫入資料對於 Transform 來說尤其成問題，因為 Transform 串流預設處於暫停狀態，直到它們被導流 (piped) 或新增了 'data' 或 'readable' 事件處理程式。

如果待寫入的資料可以按需產生或獲取，建議將邏輯封裝到 Readable 中並使用 stream.pipe()。但是，如果偏好呼叫 write()，可以使用 'drain' 事件來尊重反壓 (backpressure) 並避免記憶體問題：

function write(data, cb) {
  if (!stream.write(data)) {
    stream.once('drain', cb);
  } else {
    process.nextTick(cb);
  }
}

// Wait for cb to be called before doing any other write.
write('hello', () => {
  console.log('Write completed, do more writes now.');
});
copy

物件模式下的 Writable 串流將始終忽略 encoding 參數。

可讀串流 (Readable streams)#

可讀串流是消耗資料之*來源*的抽象化。

Readable 串流的例子包括：

• 用戶端上的 HTTP 回應
• 伺服器上的 HTTP 請求
• fs 讀取串流
• zlib 串流
• crypto 串流
• TCP sockets
• 子程序 stdout 與 stderr
• process.stdin

所有 Readable 串流都實作了由 stream.Readable 類別定義的介面。

兩種讀取模式#

Readable 串流實際上在以下兩種模式之一運作：流動模式 (flowing) 與暫停模式 (paused)。這些模式與物件模式無關。不論是否處於流動模式或暫停模式，Readable 串流都可以是或不是物件模式。

• 在流動模式下，資料會自動從底層系統讀取，並透過 EventEmitter 介面的事件儘快提供給應用程式。

• 在暫停模式下，必須明確呼叫 stream.read() 方法來從串流中讀取資料區塊。

所有 Readable 串流最初都處於暫停模式，但可以透過以下方式之一切換到流動模式：

• 新增 'data' 事件處理程式。
• 呼叫 stream.resume() 方法。
• 呼叫 stream.pipe() 方法將資料傳送到 Writable。

Readable 可以使用以下方式之一切換回暫停模式：

• 如果沒有導流目標，則呼叫 stream.pause() 方法。
• 如果有導流目標，則移除所有導流目標。可以透過呼叫 stream.unpipe() 方法移除多個導流目標。

要記住的重要概念是，除非提供了消耗或忽略資料的機制，否則 Readable 不會產生資料。如果消耗機制被停用或移除，Readable 將*嘗試*停止產生資料。

基於回溯相容性，移除 'data' 事件處理程式**不會**自動暫停串流。此外，如果有導流目標，呼叫 stream.pause() 並不能保證串流在這些目標消耗完畢並請求更多資料時會*保持*暫停狀態。

如果 Readable 切換到流動模式，但沒有可用的使用者來處理資料，則該資料將會丟失。例如，當呼叫 readable.resume() 方法但未附加 'data' 事件接聽器時，或從串流中移除 'data' 事件處理程式時，就會發生這種情況。

新增 'readable' 事件處理程式會自動使串流停止流動，資料必須透過 readable.read() 消耗。如果移除了 'readable' 事件處理程式，且有 'data' 事件處理程式，則串流將再次開始流動。

三種狀態#

Readable 串流的「兩種模式」是對 Readable 串流實作內部發生的更複雜狀態管理的簡化抽象。

具體而言，在任何給定時間點，每個 Readable 都處於以下三種可能狀態之一：

• readable.readableFlowing === null
• readable.readableFlowing === false
• readable.readableFlowing === true

當 readable.readableFlowing 為 null 時，不提供消耗串流資料的機制。因此，串流不會產生資料。在此狀態下，附加 'data' 事件的接聽器、呼叫 readable.pipe() 方法或呼叫 readable.resume() 方法將使 readable.readableFlowing 切換為 true，導致 Readable 在產生資料時開始積極地發出事件。

呼叫 readable.pause()、readable.unpipe() 或接收到反壓將導致 readable.readableFlowing 被設定為 false，暫時停止事件流，但*不*停止資料產生。在此狀態下，附加 'data' 事件的接聽器不會將 readable.readableFlowing 切換為 true。

const { PassThrough, Writable } = require('node:stream');
const pass = new PassThrough();
const writable = new Writable();

pass.pipe(writable);
pass.unpipe(writable);
// readableFlowing is now false.

pass.on('data', (chunk) => { console.log(chunk.toString()); });
// readableFlowing is still false.
pass.write('ok');  // Will not emit 'data'.
pass.resume();     // Must be called to make stream emit 'data'.
// readableFlowing is now true.
copy

當 readable.readableFlowing 為 false 時，資料可能會在串流的內部緩衝區中累積。

選擇一種 API 風格#

Readable 串流 API 跨多個 Node.js 版本演進，並提供多種消耗串流資料的方法。通常，開發人員應選擇*其中一種*消耗資料的方法，*絕不應*使用多種方法從單個串流消耗資料。具體而言，結合使用 on('data')、on('readable')、pipe() 或非同步迭代器可能會導致不直覺的行為。

類別：stream.Readable#

事件：'close'#

當串流及其任何底層資源（例如檔案描述符）已關閉時，會發出 'close' 事件。該事件表示將不再發出任何事件，也不會再進行任何計算。

如果 Readable 串流是在帶有 emitClose 選項的情況下建立的，則它始終會發出 'close' 事件。

事件：'data'#

• chunk <Buffer> | <string> | <any> 資料區塊。對於不在物件模式下運作的串流，區塊將是字串或 Buffer。對於物件模式串流，區塊可以是除 null 以外的任何 JavaScript 值。

每當串流將資料區塊的所有權移交給使用者時，都會發出 'data' 事件。每當透過呼叫 readable.pipe()、readable.resume() 或透過為 'data' 事件附加接聽器回呼而將串流切換到流動模式時，都可能發生這種情況。每當呼叫 readable.read() 方法且有資料區塊可回傳時，也會發出 'data' 事件。

為未明確暫停的串流附加 'data' 事件接聽器會將串流切換到流動模式。資料將在可用時立即傳遞。

如果使用 readable.setEncoding() 方法為串流指定了預設編碼，則接聽器回呼將接收到字串形式的資料區塊；否則資料將以 Buffer 形式傳遞。

const readable = getReadableStreamSomehow();
readable.on('data', (chunk) => {
  console.log(`Received ${chunk.length} bytes of data.`);
});
copy

事件：'end'#

當串流中不再有資料可供消耗時，會發出 'end' 事件。

除非資料已被完全消耗，否則 'end' 事件**不會發出**。這可以透過將串流切換到流動模式，或重複呼叫 stream.read() 直到所有資料都被消耗來達成。

const readable = getReadableStreamSomehow();
readable.on('data', (chunk) => {
  console.log(`Received ${chunk.length} bytes of data.`);
});
readable.on('end', () => {
  console.log('There will be no more data.');
});
copy

事件：'error'#

• 類型：<Error>

Readable 實作可能隨時發出 'error' 事件。通常，如果底層串流由於底層內部故障而無法產生資料，或者當串流實作嘗試推送無效的資料區塊時，就會發生這種情況。

接聽器回呼將接收到單個 Error 物件。

事件：'pause'#

當呼叫 stream.pause() 且 readableFlowing 不為 false 時，會發出 'pause' 事件。

事件：'readable'#

當串流中有資料可供讀取（最多到配置的高水位線 state.highWaterMark）時，會發出 'readable' 事件。實際上，它表示串流在緩衝區內有新資訊。如果緩衝區內有資料，可以呼叫 stream.read() 來檢索該資料。此外，當達到串流結尾時，也可能發出 'readable' 事件。

const readable = getReadableStreamSomehow();
readable.on('readable', function() {
  // There is some data to read now.
  let data;

  while ((data = this.read()) !== null) {
    console.log(data);
  }
});
copy

如果已達到串流結尾，呼叫 stream.read() 將回傳 null 並觸發 'end' 事件。如果從未有任何資料可讀，情況也是如此。例如，在下例中，foo.txt 是一個空檔案：

const fs = require('node:fs');
const rr = fs.createReadStream('foo.txt');
rr.on('readable', () => {
  console.log(`readable: ${rr.read()}`);
});
rr.on('end', () => {
  console.log('end');
});
copy

執行此腳本的輸出為：

$ node test.js
readable: null
end
copy

在某些情況下，為 'readable' 事件附加接聽器會導致一定數量的資料被讀入內部緩衝區。

通常，readable.pipe() 與 'data' 事件機制比 'readable' 事件更容易理解。然而，處理 'readable' 可能會提高吞吐量。

如果同時使用 'readable' 與 'data'，則 'readable' 在控制流動方面具有優先權，即 'data' 僅在呼叫 stream.read() 時發出。readableFlowing 屬性將變為 false。如果在移除 'readable' 時有 'data' 接聽器，串流將開始流動，即無需呼叫 .resume() 即可發出 'data' 事件。

事件：'resume'#

當呼叫 stream.resume() 且 readableFlowing 不為 true 時，會發出 'resume' 事件。

readable.destroy([error])#

• error <Error> 將作為 'error' 事件負載傳遞的錯誤
• 傳回：<this>

銷毀串流。選配地發出 'error' 事件，並發出 'close' 事件（除非 emitClose 設定為 false）。在此呼叫之後，可讀串流將釋放任何內部資源，隨後對 push() 的呼叫將被忽略。

一旦呼叫了 destroy()，任何進一步的呼叫都將是空操作，除了來自 _destroy() 的錯誤外，不會再發出進一步的 'error'。

實作編寫者不應覆寫此方法，而應實作 readable._destroy()。

readable.closed#

• 類型：<boolean>

在 'close' 發出後為 true。

readable.destroyed#

• 類型：<boolean>

在呼叫 readable.destroy() 後為 true。

readable.isPaused()#

• 傳回：<boolean>

readable.isPaused() 方法回傳 Readable 目前的運作狀態。這主要由 readable.pipe() 方法底層的機制使用。在大多數典型情況下，沒有理由直接使用此方法。

const readable = new stream.Readable();

readable.isPaused(); // === false
readable.pause();
readable.isPaused(); // === true
readable.resume();
readable.isPaused(); // === false
copy

readable.pause()#

• 傳回：<this>

readable.pause() 方法將使處於流動模式的串流停止發出 'data' 事件，並切換出流動模式。任何變為可用的資料都將保留在內部緩衝區中。

const readable = getReadableStreamSomehow();
readable.on('data', (chunk) => {
  console.log(`Received ${chunk.length} bytes of data.`);
  readable.pause();
  console.log('There will be no additional data for 1 second.');
  setTimeout(() => {
    console.log('Now data will start flowing again.');
    readable.resume();
  }, 1000);
});
copy

如果存在 'readable' 事件接聽器，readable.pause() 方法將不起作用。

readable.pipe(destination[, options])#

• destination <stream.Writable> 資料寫入的目標
• options <Object> 導流選項
  • end <boolean> 當讀取端結束時結束寫入端。預設值： true。
• 回傳：<stream.Writable> 目標 (*destination*)，如果它是 Duplex 或 Transform 串流，則允許鏈式導流

readable.pipe() 方法將 Writable 串流連接到 readable，使其自動切換到流動模式並將其所有資料推送到連接的 Writable。資料流將自動管理，以便目標 Writable 串流不會被更快的 Readable 串流淹沒。

以下範例將 readable 的所有資料導流到名為 file.txt 的檔案中：

const fs = require('node:fs');
const readable = getReadableStreamSomehow();
const writable = fs.createWriteStream('file.txt');
// All the data from readable goes into 'file.txt'.
readable.pipe(writable);
copy

可以將多個 Writable 串流連接到單個 Readable 串流。

readable.pipe() 方法回傳對目標串流的引用，使得建立導流串流鏈成為可能：

const fs = require('node:fs');
const zlib = require('node:zlib');
const r = fs.createReadStream('file.txt');
const z = zlib.createGzip();
const w = fs.createWriteStream('file.txt.gz');
r.pipe(z).pipe(w);
copy

預設情況下，當來源 Readable 串流發出 'end' 時，會在目標 Writable 串流上呼叫 stream.end()，使得目標不再可寫。若要停用此預設行為，可以將 end 選項傳遞為 false，讓目標串流保持開啟：

reader.pipe(writer, { end: false });
reader.on('end', () => {
  writer.end('Goodbye\n');
});
copy

一個重要的警告是，如果 Readable 串流在處理過程中發出錯誤，目標 Writable *不會*自動關閉。如果發生錯誤，則需要*手動*關閉每個串流以防止記憶體洩漏。

無論指定的選項為何，process.stderr 與 process.stdout Writable 串流直到 Node.js 程序結束前都不會關閉。

readable.read([size])#

• size <number> 選配參數，指定要讀取的資料量。
• 回傳：<string> | <Buffer> | <null> | <any>

readable.read() 方法從內部緩衝區中讀取資料並將其回傳。如果沒有資料可讀，則回傳 null。預設情況下，資料以 Buffer 物件形式回傳，除非已使用 readable.setEncoding() 方法指定編碼或串流運作於物件模式。

選配的 size 參數指定要讀取的特定位元組數。如果沒有 size 位元組的資料可讀，則將回傳 null，*除非*串流已結束，在這種情況下將回傳內部緩衝區中剩餘的所有資料。

如果未指定 size 參數，則將回傳內部緩衝區中包含的所有資料。

size 參數必須小於或等於 1 GiB。

readable.read() 方法應僅在處於暫停模式的 Readable 串流上呼叫。在流動模式下，readable.read() 會自動被呼叫，直到內部緩衝區完全消耗完畢。

const readable = getReadableStreamSomehow();

// 'readable' may be triggered multiple times as data is buffered in
readable.on('readable', () => {
  let chunk;
  console.log('Stream is readable (new data received in buffer)');
  // Use a loop to make sure we read all currently available data
  while (null !== (chunk = readable.read())) {
    console.log(`Read ${chunk.length} bytes of data...`);
  }
});

// 'end' will be triggered once when there is no more data available
readable.on('end', () => {
  console.log('Reached end of stream.');
});
copy

每次對 readable.read() 的呼叫都會回傳一個資料區塊或 null，表示目前沒有更多資料可讀。這些區塊不會自動串接。由於單次 read() 呼叫不會回傳所有資料，因此可能需要使用 while 迴圈持續讀取區塊，直到檢索到所有資料。讀取大檔案時，.read() 可能會暫時回傳 null，表示它已消耗完所有緩衝內容，但可能還有更多資料待緩衝。在這種情況下，一旦緩衝區中有更多資料，就會發出新的 'readable' 事件，而 'end' 事件表示資料傳輸結束。

因此，若要從 readable 讀取檔案的全部內容，必須在多個 'readable' 事件中收集資料區塊：

const chunks = [];

readable.on('readable', () => {
  let chunk;
  while (null !== (chunk = readable.read())) {
    chunks.push(chunk);
  }
});

readable.on('end', () => {
  const content = chunks.join('');
});
copy

不論 size 參數的值為何，物件模式下的 Readable 串流呼叫 readable.read(size) 始終會回傳單個項目。

如果 readable.read() 方法回傳一個資料區塊，也會發出 'data' 事件。

在發出 'end' 事件後呼叫 stream.read([size]) 將回傳 null。不會引發執行階段錯誤。

readable.readable#

• 類型：<boolean>

如果呼叫 readable.read() 是安全的（表示串流尚未被銷毀、報錯或結束），則為 true。

readable.readableAborted#

• 類型：<boolean>

回傳串流是否在發出 'end' 之前已被銷毀或報錯。

readable.readableDidRead#

• 類型：<boolean>

回傳是否已發出 'data'。

readable.readableEncoding#

• 類型：<null> | <string>

給定 Readable 串流之 encoding 屬性的 getter。可以使用 readable.setEncoding() 方法設定 encoding 屬性。

readable.readableEnded#

• 類型：<boolean>

當發出 'end' 事件時變為 true。

readable.errored#

• 類型：<Error>

如果串流在銷毀時帶有錯誤，則回傳該錯誤。

readable.readableFlowing#

• 類型：<boolean>

此屬性反映了 三種狀態 章節中描述的 Readable 串流目前狀態。

readable.readableHighWaterMark#

• 類型：<number>

回傳建立此 Readable 時傳入的 highWaterMark 值。

readable.readableLength#

• 類型：<number>

此屬性包含隊列中準備讀取的位元組數（或物件數）。此值提供了關於 highWaterMark 狀態的內省資料。

readable.readableObjectMode#

• 類型：<boolean>

給定 Readable 串流之 objectMode 屬性的 getter。

readable.resume()#

• 傳回：<this>

readable.resume() 方法使明確暫停的 Readable 串流恢復發出 'data' 事件，並將串流切換到流動模式。

readable.resume() 方法可以用於完全消耗來自串流的資料，而無需實際處理任何資料：

getReadableStreamSomehow()
  .resume()
  .on('end', () => {
    console.log('Reached the end, but did not read anything.');
  });
copy

如果存在 'readable' 事件接聽器，readable.resume() 方法將不起作用。

readable.setEncoding(encoding)#

• encoding <string> 要使用的編碼。
• 傳回：<this>

readable.setEncoding() 方法設定從 Readable 串流讀取之資料的字元編碼。

預設情況下，未分配編碼，串流資料將以 Buffer 物件形式回傳。設定編碼會導致串流資料以指定編碼的字串而非 Buffer 物件形式回傳。例如，呼叫 readable.setEncoding('utf8') 將導致輸出資料被解釋為 UTF-8 資料並作為字串傳遞。呼叫 readable.setEncoding('hex') 將使資料以十六進位字串格式編碼。

Readable 串流將正確處理透過串流傳遞的多位元組字元，如果僅將其作為 Buffer 物件從串流中拉取，這些字元可能會被錯誤地解碼。

const readable = getReadableStreamSomehow();
readable.setEncoding('utf8');
readable.on('data', (chunk) => {
  assert.equal(typeof chunk, 'string');
  console.log('Got %d characters of string data:', chunk.length);
});
copy

readable.unpipe([destination])#

• destination <stream.Writable> 選配，要取消導流的特定串流
• 傳回：<this>

readable.unpipe() 方法分離先前使用 stream.pipe() 方法附加的 Writable 串流。

如果未指定 destination，則分離*所有*管道。

如果指定了 destination，但未對其設置管道，則該方法不執行任何操作。

const fs = require('node:fs');
const readable = getReadableStreamSomehow();
const writable = fs.createWriteStream('file.txt');
// All the data from readable goes into 'file.txt',
// but only for the first second.
readable.pipe(writable);
setTimeout(() => {
  console.log('Stop writing to file.txt.');
  readable.unpipe(writable);
  console.log('Manually close the file stream.');
  writable.end();
}, 1000);
copy

readable.unshift(chunk[, encoding])#

• chunk <Buffer> | <TypedArray> | <DataView> | <string> | <null> | <any> 推回讀取隊列的資料區塊。對於不在物件模式下運作的串流，chunk 必須是 <string>, <Buffer>, <TypedArray>, <DataView> 或 null。對於物件模式串流，chunk 可以是任何 JavaScript 值。
• encoding <string> 字串區塊的編碼。必須是有效的 Buffer 編碼，例如 'utf8' 或 'ascii'。

將 chunk 作為 null 傳遞表示串流結束 (EOF)，其行為與 readable.push(null) 相同，之後無法再寫入任何資料。EOF 信號被置於緩衝區末尾，任何已緩衝的資料仍將被排空。

readable.unshift() 方法將資料區塊推回內部緩衝區。這在某些情況下很有用，例如當程式碼消耗串流時，需要「退回」一些它樂觀地從來源拉取的資料，以便將資料傳遞給其他方。

在發出 'end' 事件後不能呼叫 stream.unshift(chunk) 方法，否則將拋出執行階段錯誤。

使用 stream.unshift() 的開發者通常應考慮改用 Transform 串流。更多資訊請參閱 串流實作者 API 章節。

// Pull off a header delimited by \n\n.
// Use unshift() if we get too much.
// Call the callback with (error, header, stream).
const { StringDecoder } = require('node:string_decoder');
function parseHeader(stream, callback) {
  stream.on('error', callback);
  stream.on('readable', onReadable);
  const decoder = new StringDecoder('utf8');
  let header = '';
  function onReadable() {
    let chunk;
    while (null !== (chunk = stream.read())) {
      const str = decoder.write(chunk);
      if (str.includes('\n\n')) {
        // Found the header boundary.
        const split = str.split(/\n\n/);
        header += split.shift();
        const remaining = split.join('\n\n');
        const buf = Buffer.from(remaining, 'utf8');
        stream.removeListener('error', callback);
        // Remove the 'readable' listener before unshifting.
        stream.removeListener('readable', onReadable);
        if (buf.length)
          stream.unshift(buf);
        // Now the body of the message can be read from the stream.
        callback(null, header, stream);
        return;
      }
      // Still reading the header.
      header += str;
    }
  }
}
copy

與 stream.push(chunk) 不同，stream.unshift(chunk) 不會透過重設串流內部的讀取狀態來結束讀取程序。如果在讀取過程中（即從自訂串流的 stream._read() 實作內部）呼叫 readable.unshift()，可能會導致非預期的結果。在呼叫 readable.unshift() 後立即呼叫 stream.push('') 會適當地重設讀取狀態，然而，最好避免在執行讀取的過程中呼叫 readable.unshift()。

readable.wrap(stream)#

• stream <Stream> 一個「舊式」的可讀串流
• 傳回：<this>

在 Node.js 0.10 之前，串流並未實作目前定義的整個 node:stream 模組 API。（更多資訊請參閱 相容性。）

當使用會發出 'data' 事件且具有僅具建議性的 stream.pause() 方法的舊版 Node.js 函式庫時，可以使用 readable.wrap() 方法建立一個使用該舊串流作為其資料來源的 Readable 串流。

很少需要使用 readable.wrap()，但該方法是為了方便與舊版 Node.js 應用程式和函式庫進行互動而提供的。

const { OldReader } = require('./old-api-module.js');
const { Readable } = require('node:stream');
const oreader = new OldReader();
const myReader = new Readable().wrap(oreader);

myReader.on('readable', () => {
  myReader.read(); // etc.
});
copy

readable[Symbol.asyncIterator]()#

• 回傳：<AsyncIterator> 用於完全消耗串流。

const fs = require('node:fs');

async function print(readable) {
  readable.setEncoding('utf8');
  let data = '';
  for await (const chunk of readable) {
    data += chunk;
  }
  console.log(data);
}

print(fs.createReadStream('file')).catch(console.error);
copy

如果迴圈以 break、return 或 throw 終止，串流將被銷毀。換句話說，迭代串流將完全消耗該串流。串流將以等於 highWaterMark 選項大小的區塊進行讀取。在上面的程式碼範例中，如果檔案的資料少於 64 KiB，則資料將在單個區塊中，因為沒有為 fs.createReadStream() 提供 highWaterMark 選項。

readable[Symbol.asyncDispose]()#

使用 AbortError 呼叫 readable.destroy() 並回傳一個在串流結束時完成 (fulfill) 的 promise。

readable.compose(stream[, options])#

• stream <Writable> | <Duplex> | <WritableStream> | <TransformStream> | <Function>
• options <Object>
  • signal <AbortSignal> 如果信號被中止，允許銷毀串流。
• 回傳：<Duplex> 與 stream 組合的串流。

import { Readable } from 'node:stream';

async function* splitToWords(source) {
  for await (const chunk of source) {
    const words = String(chunk).split(' ');

    for (const word of words) {
      yield word;
    }
  }
}

const wordsStream = Readable.from(['text passed through', 'composed stream']).compose(splitToWords);
const words = await wordsStream.toArray();

console.log(words); // prints ['text', 'passed', 'through', 'composed', 'stream']
copy

readable.compose(s) 等同於 stream.compose(readable, s)。

此方法還允許提供 <AbortSignal>，當中止時會銷毀組合後的串流。

更多資訊請參閱 stream.compose(...streams)。

readable.iterator([options])#

• options <Object>
  • destroyOnReturn <boolean> 當設定為 false 時，在非同步迭代器上呼叫 return，或使用 break、return 或 throw 結束 for await...of 迭代將不會銷毀串流。預設值： true。
• 回傳：<AsyncIterator> 用於消耗串流。

此方法建立的迭代器讓使用者可以選擇在 for await...of 迴圈透過 return、break 或 throw 結束時，是否取消串流的銷毀，或者在迭代期間串流發出錯誤時，迭代器是否應該銷毀串流。

const { Readable } = require('node:stream');

async function printIterator(readable) {
  for await (const chunk of readable.iterator({ destroyOnReturn: false })) {
    console.log(chunk); // 1
    break;
  }

  console.log(readable.destroyed); // false

  for await (const chunk of readable.iterator({ destroyOnReturn: false })) {
    console.log(chunk); // Will print 2 and then 3
  }

  console.log(readable.destroyed); // True, stream was totally consumed
}

async function printSymbolAsyncIterator(readable) {
  for await (const chunk of readable) {
    console.log(chunk); // 1
    break;
  }

  console.log(readable.destroyed); // true
}

async function showBoth() {
  await printIterator(Readable.from([1, 2, 3]));
  await printSymbolAsyncIterator(Readable.from([1, 2, 3]));
}

showBoth();
copy

readable.map(fn[, options])#

• fn <Function> | <AsyncFunction> 用於對串流中的每個區塊進行映射 (map) 的函數。
  • data <any> 來自串流的一個資料區塊。
  • options <Object>
    • signal <AbortSignal> 如果串流被銷毀則中止，允許提前中止 fn 呼叫。
• options <Object>
  • concurrency <number> 串流上一次可以同時呼叫 fn 的最大並發數。預設值： 1。
  • highWaterMark <number> 在等待使用者消耗已映射項目時要緩衝多少個項目。預設值： concurrency * 2 - 1。
  • signal <AbortSignal> 如果信號被中止，允許銷毀串流。
• 回傳：<Readable> 使用函數 fn 映射後的串流。

此方法允許對串流進行映射。對串流中的每個區塊都會呼叫一次 fn 函數。如果 fn 函數回傳一個 promise，則在將結果傳遞給結果串流之前會先 await 該 promise。

import { Readable } from 'node:stream';
import { Resolver } from 'node:dns/promises';

// With a synchronous mapper.
for await (const chunk of Readable.from([1, 2, 3, 4]).map((x) => x * 2)) {
  console.log(chunk); // 2, 4, 6, 8
}
// With an asynchronous mapper, making at most 2 queries at a time.
const resolver = new Resolver();
const dnsResults = Readable.from([
  'nodejs.org',
  'openjsf.org',
  'www.linuxfoundation.org',
]).map((domain) => resolver.resolve4(domain), { concurrency: 2 });
for await (const result of dnsResults) {
  console.log(result); // Logs the DNS result of resolver.resolve4.
}
copy

readable.filter(fn[, options])#

• fn <Function> | <AsyncFunction> 用於過濾串流區塊的函數。
  • data <any> 來自串流的一個資料區塊。
  • options <Object>
    • signal <AbortSignal> 如果串流被銷毀則中止，允許提前中止 fn 呼叫。
• options <Object>
  • concurrency <number> 串流上一次可以同時呼叫 fn 的最大並發數。預設值： 1。
  • highWaterMark <number> 在等待使用者消耗已過濾項目時要緩衝多少個項目。預設值： concurrency * 2 - 1。
  • signal <AbortSignal> 如果信號被中止，允許銷毀串流。
• 回傳：<Readable> 使用謂詞 fn 過濾後的串流。

此方法允許過濾串流。對於串流中的每個區塊，都會呼叫 fn 函數，如果它回傳真值 (truthy)，則該區塊將被傳遞給結果串流。如果 fn 函數回傳一個 promise，則會先 await 該 promise。

import { Readable } from 'node:stream';
import { Resolver } from 'node:dns/promises';

// With a synchronous predicate.
for await (const chunk of Readable.from([1, 2, 3, 4]).filter((x) => x > 2)) {
  console.log(chunk); // 3, 4
}
// With an asynchronous predicate, making at most 2 queries at a time.
const resolver = new Resolver();
const dnsResults = Readable.from([
  'nodejs.org',
  'openjsf.org',
  'www.linuxfoundation.org',
]).filter(async (domain) => {
  const { address } = await resolver.resolve4(domain, { ttl: true });
  return address.ttl > 60;
}, { concurrency: 2 });
for await (const result of dnsResults) {
  // Logs domains with more than 60 seconds on the resolved dns record.
  console.log(result);
}
copy

readable.forEach(fn[, options])#

• fn <Function> | <AsyncFunction> 對串流的每個區塊呼叫的函數。
  • data <any> 來自串流的一個資料區塊。
  • options <Object>
    • signal <AbortSignal> 如果串流被銷毀則中止，允許提前中止 fn 呼叫。
• options <Object>
  • concurrency <number> 串流上一次可以同時呼叫 fn 的最大並發數。預設值： 1。
  • signal <AbortSignal> 如果信號被中止，允許銷毀串流。
• 回傳：<Promise> 串流完成時的 promise。

此方法允許迭代串流。對於串流中的每個區塊，都會呼叫 fn 函數。如果 fn 函數回傳一個 promise，則會先 await 該 promise。

此方法與 for await...of 迴圈不同之處在於它可以選擇性地並發處理區塊。此外，forEach 迭代只能透過傳遞 signal 選項並中止相關的 AbortController 來停止，而 for await...of 可以使用 break 或 return 停止。在任何一種情況下，串流都將被銷毀。

此方法與監聽 'data' 事件不同之處在於它在底層機制中使用了 readable 事件，並且可以限制並發 fn 呼叫的數量。

import { Readable } from 'node:stream';
import { Resolver } from 'node:dns/promises';

// With a synchronous predicate.
for await (const chunk of Readable.from([1, 2, 3, 4]).filter((x) => x > 2)) {
  console.log(chunk); // 3, 4
}
// With an asynchronous predicate, making at most 2 queries at a time.
const resolver = new Resolver();
const dnsResults = Readable.from([
  'nodejs.org',
  'openjsf.org',
  'www.linuxfoundation.org',
]).map(async (domain) => {
  const { address } = await resolver.resolve4(domain, { ttl: true });
  return address;
}, { concurrency: 2 });
await dnsResults.forEach((result) => {
  // Logs result, similar to `for await (const result of dnsResults)`
  console.log(result);
});
console.log('done'); // Stream has finished
copy

readable.toArray([options])#

• options <Object>
  • signal <AbortSignal> 如果信號被中止，允許取消 toArray 操作。
• 回傳：<Promise> 一個包含串流內容陣列的 promise。

此方法可以輕鬆獲取串流的內容。

由於此方法會將整個串流讀入記憶體，因此它抵消了串流的優點。它旨在用於互操作性和便利性，而不是作為消耗串流的主要方式。

import { Readable } from 'node:stream';
import { Resolver } from 'node:dns/promises';

await Readable.from([1, 2, 3, 4]).toArray(); // [1, 2, 3, 4]

const resolver = new Resolver();

// Make dns queries concurrently using .map and collect
// the results into an array using toArray
const dnsResults = await Readable.from([
  'nodejs.org',
  'openjsf.org',
  'www.linuxfoundation.org',
]).map(async (domain) => {
  const { address } = await resolver.resolve4(domain, { ttl: true });
  return address;
}, { concurrency: 2 }).toArray();
copy

readable.some(fn[, options])#

• fn <Function> | <AsyncFunction> 對串流的每個區塊呼叫的函數。
  • data <any> 來自串流的一個資料區塊。
  • options <Object>
    • signal <AbortSignal> 如果串流被銷毀則中止，允許提前中止 fn 呼叫。
• options <Object>
  • concurrency <number> 串流上一次可以同時呼叫 fn 的最大並發數。預設值： 1。
  • signal <AbortSignal> 如果信號被中止，允許銷毀串流。
• 回傳：<Promise> 如果 fn 對於至少一個區塊回傳真值，則此 promise 的評估結果為 true。

此方法類似於 Array.prototype.some，並對串流中的每個區塊呼叫 fn，直到 awaited 的回傳值為 true（或任何真值）。一旦某個區塊的 fn 呼叫 awaited 回傳值為真，串流將被銷毀，並且 promise 將以 true 完成。如果對所有區塊的 fn 呼叫都沒有回傳真值，則 promise 將以 false 完成。

import { Readable } from 'node:stream';
import { stat } from 'node:fs/promises';

// With a synchronous predicate.
await Readable.from([1, 2, 3, 4]).some((x) => x > 2); // true
await Readable.from([1, 2, 3, 4]).some((x) => x < 0); // false

// With an asynchronous predicate, making at most 2 file checks at a time.
const anyBigFile = await Readable.from([
  'file1',
  'file2',
  'file3',
]).some(async (fileName) => {
  const stats = await stat(fileName);
  return stats.size > 1024 * 1024;
}, { concurrency: 2 });
console.log(anyBigFile); // `true` if any file in the list is bigger than 1MB
console.log('done'); // Stream has finished
copy

readable.find(fn[, options])#

• fn <Function> | <AsyncFunction> 對串流的每個區塊呼叫的函數。
  • data <any> 來自串流的一個資料區塊。
  • options <Object>
    • signal <AbortSignal> 如果串流被銷毀則中止，允許提前中止 fn 呼叫。
• options <Object>
  • concurrency <number> 串流上一次可以同時呼叫 fn 的最大並發數。預設值： 1。
  • signal <AbortSignal> 如果信號被中止，允許銷毀串流。
• 回傳：<Promise> 此 promise 的評估結果為 fn 評估為真值的第一個區塊，如果未找到任何元素，則為 undefined。

此方法類似於 Array.prototype.find，並對串流中的每個區塊呼叫 fn，以尋找使 fn 為真值的區塊。一旦某個 fn 呼叫的 awaited 回傳值為真，串流將被銷毀，並且 promise 將以 fn 回傳真值的該值完成。如果對所有區塊的 fn 呼叫都回傳假值 (falsy)，則 promise 將以 undefined 完成。

import { Readable } from 'node:stream';
import { stat } from 'node:fs/promises';

// With a synchronous predicate.
await Readable.from([1, 2, 3, 4]).find((x) => x > 2); // 3
await Readable.from([1, 2, 3, 4]).find((x) => x > 0); // 1
await Readable.from([1, 2, 3, 4]).find((x) => x > 10); // undefined

// With an asynchronous predicate, making at most 2 file checks at a time.
const foundBigFile = await Readable.from([
  'file1',
  'file2',
  'file3',
]).find(async (fileName) => {
  const stats = await stat(fileName);
  return stats.size > 1024 * 1024;
}, { concurrency: 2 });
console.log(foundBigFile); // File name of large file, if any file in the list is bigger than 1MB
console.log('done'); // Stream has finished
copy

readable.every(fn[, options])#

• fn <Function> | <AsyncFunction> 對串流的每個區塊呼叫的函數。
  • data <any> 來自串流的一個資料區塊。
  • options <Object>
    • signal <AbortSignal> 如果串流被銷毀則中止，允許提前中止 fn 呼叫。
• options <Object>
  • concurrency <number> 串流上一次可以同時呼叫 fn 的最大並發數。預設值： 1。
  • signal <AbortSignal> 如果信號被中止，允許銷毀串流。
• 回傳：<Promise> 如果 fn 對所有區塊都回傳真值，則此 promise 的評估結果為 true。

此方法類似於 Array.prototype.every，並對串流中的每個區塊呼叫 fn，以檢查是否所有 awaited 回傳值對於 fn 都是真值。一旦某個區塊的 fn 呼叫 awaited 回傳值為假，串流將被銷毀，並且 promise 將以 false 完成。如果對所有區塊的 fn 呼叫都回傳真值，則 promise 將以 true 完成。

import { Readable } from 'node:stream';
import { stat } from 'node:fs/promises';

// With a synchronous predicate.
await Readable.from([1, 2, 3, 4]).every((x) => x > 2); // false
await Readable.from([1, 2, 3, 4]).every((x) => x > 0); // true

// With an asynchronous predicate, making at most 2 file checks at a time.
const allBigFiles = await Readable.from([
  'file1',
  'file2',
  'file3',
]).every(async (fileName) => {
  const stats = await stat(fileName);
  return stats.size > 1024 * 1024;
}, { concurrency: 2 });
// `true` if all files in the list are bigger than 1MiB
console.log(allBigFiles);
console.log('done'); // Stream has finished
copy

readable.flatMap(fn[, options])#

• fn <Function> | <AsyncGeneratorFunction> | <AsyncFunction> 用於對串流中的每個區塊進行映射的函數。
  • data <any> 來自串流的一個資料區塊。
  • options <Object>
    • signal <AbortSignal> 如果串流被銷毀則中止，允許提前中止 fn 呼叫。
• options <Object>
  • concurrency <number> 串流上一次可以同時呼叫 fn 的最大並發數。預設值： 1。
  • signal <AbortSignal> 如果信號被中止，允許銷毀串流。
• 回傳：<Readable> 使用函數 fn 進行扁平映射 (flat-mapped) 後的串流。

此方法透過對串流的每個區塊應用給定的回呼函數，然後將結果扁平化，從而回傳一個新串流。

可以從 fn 回傳一個串流或其他可迭代對象或非同步可迭代對象，結果串流將合併（扁平化）到回傳的串流中。

import { Readable } from 'node:stream';
import { createReadStream } from 'node:fs';

// With a synchronous mapper.
for await (const chunk of Readable.from([1, 2, 3, 4]).flatMap((x) => [x, x])) {
  console.log(chunk); // 1, 1, 2, 2, 3, 3, 4, 4
}
// With an asynchronous mapper, combine the contents of 4 files
const concatResult = Readable.from([
  './1.mjs',
  './2.mjs',
  './3.mjs',
  './4.mjs',
]).flatMap((fileName) => createReadStream(fileName));
for await (const result of concatResult) {
  // This will contain the contents (all chunks) of all 4 files
  console.log(result);
}
copy

readable.drop(limit[, options])#

• limit <number> 要從可讀串流中捨棄的區塊數量。
• options <Object>
  • signal <AbortSignal> 如果信號被中止，允許銷毀串流。
• 回傳：<Readable> 捨棄了 limit 個區塊後的串流。

此方法回傳一個捨棄了前 limit 個區塊的新串流。

import { Readable } from 'node:stream';

await Readable.from([1, 2, 3, 4]).drop(2).toArray(); // [3, 4]
copy

readable.take(limit[, options])#

• limit <number> 要從可讀串流中獲取的區塊數量。
• options <Object>
  • signal <AbortSignal> 如果信號被中止，允許銷毀串流。
• 回傳：<Readable> 獲取了 limit 個區塊後的串流。

此方法回傳一個包含前 limit 個區塊的新串流。

import { Readable } from 'node:stream';

await Readable.from([1, 2, 3, 4]).take(2).toArray(); // [1, 2]
copy

readable.reduce(fn[, initial[, options]])#

• fn <Function> | <AsyncFunction> 用於對串流中的每個區塊呼叫的歸納 (reducer) 函數。
  • previous <any> 從上一次呼叫 fn 獲得的值，或指定的 initial 值，否則為串流的第一個區塊。
  • data <any> 來自串流的一個資料區塊。
  • options <Object>
    • signal <AbortSignal> 如果串流被銷毀則中止，允許提前中止 fn 呼叫。
• initial <any> 歸納計算中使用的初始值。
• options <Object>
  • signal <AbortSignal> 如果信號被中止，允許銷毀串流。
• 回傳：<Promise> 歸納計算最終值的 promise。

此方法按順序對串流的每個區塊呼叫 fn，並將上一個元素的計算結果傳遞給它。它會回傳一個歸納計算最終值的 promise。

如果沒有提供 initial 值，則使用串流的第一個區塊作為初始值。如果串流為空，則 promise 會被拒絕，並帶有 TypeError 和 ERR_INVALID_ARGS 錯誤代碼屬性。

import { Readable } from 'node:stream';
import { readdir, stat } from 'node:fs/promises';
import { join } from 'node:path';

const directoryPath = './src';
const filesInDir = await readdir(directoryPath);

const folderSize = await Readable.from(filesInDir)
  .reduce(async (totalSize, file) => {
    const { size } = await stat(join(directoryPath, file));
    return totalSize + size;
  }, 0);

console.log(folderSize);
copy

歸納函數逐個元素地迭代串流，這意味著沒有 concurrency 參數或並行處理。要同時執行 reduce，您可以將非同步函數提取到 readable.map 方法中。

import { Readable } from 'node:stream';
import { readdir, stat } from 'node:fs/promises';
import { join } from 'node:path';

const directoryPath = './src';
const filesInDir = await readdir(directoryPath);

const folderSize = await Readable.from(filesInDir)
  .map((file) => stat(join(directoryPath, file)), { concurrency: 2 })
  .reduce((totalSize, { size }) => totalSize + size, 0);

console.log(folderSize);
copy

雙工與轉換串流#

類別：stream.Duplex#

雙工串流是同時實作了 Readable 和 Writable 介面的串流。

Duplex 串流的範例包括

• TCP sockets
• zlib 串流
• crypto 串流

duplex.allowHalfOpen#

• 類型：<boolean>

如果為 false，則串流會在可讀端結束時自動結束可寫端。最初由 allowHalfOpen 建構函式選項設定，預設值為 true。

這可以手動更改以更改現有 Duplex 串流實例的半開放行為，但必須在發出 'end' 事件之前更改。

類別：stream.Transform#

轉換串流是輸出以某種方式與輸入相關聯的 Duplex 串流。與所有 Duplex 串流一樣，Transform 串流同時實作了 Readable 和 Writable 介面。

Transform 串流的範例包括

• zlib 串流
• crypto 串流

transform.destroy([error])#

• error <Error>
• 傳回：<this>

銷毀串流，並選擇性地發出 'error' 事件。在此呼叫之後，轉換串流將釋放任何內部資源。實作者不應覆寫此方法，而應實作 readable._destroy()。Transform 的 _destroy() 預設實作也會發出 'close'，除非 emitClose 設定為 false。

一旦呼叫了 destroy()，任何進一步的呼叫都將不起作用，除了來自 _destroy() 之外的任何進一步錯誤都不會以 'error' 的形式發出。

stream.duplexPair([options])#

• options <Object> 要傳遞給兩個 Duplex 建構函式的值，用於設定緩衝等選項。
• 回傳：由兩個 Duplex 實例組成的 <Array>。

工具函數 duplexPair 回傳一個包含兩個項目的陣列，每個項目都是一個連接到另一側的 Duplex 串流。

const [ sideA, sideB ] = duplexPair();
copy

寫入其中一個串流的任何內容都會在另一個串流中變為可讀。它提供了類似於網路連接的行為，其中用戶端寫入的資料變為伺服器可讀的資料，反之亦然。

Duplex 串流是對稱的；可以使用其中之一，行為沒有任何區別。

stream.finished(stream[, options], callback)#

• stream <Stream> | <ReadableStream> | <WritableStream> 一個可讀及/或可寫的串流/網頁串流 (webstream)。
• options <Object>
  • error <boolean> 如果設定為 false，則呼叫 emit('error', err) 不會被視為已結束。預設值： true。
  • readable <boolean> 當設定為 false 時，即使串流可能仍然可讀，當串流結束時也會呼叫回呼。預設值： true。
  • writable <boolean> 當設定為 false 時，即使串流可能仍然可寫，當串流結束時也會呼叫回呼。預設值： true。
  • signal <AbortSignal> 允許中止對串流完成的等待。如果信號被中止，底層串流將*不會*被中止。回呼將被呼叫並帶有 AbortError。此函數新增的所有已註冊監聽器也將被移除。
• callback <Function> 接受一個選用的錯誤參數的回呼函數。
• 回傳：<Function> 一個用於移除所有已註冊監聽器的清理函數。

當串流不再可讀、可寫或遇到錯誤或過早關閉事件時得到通知的函數。

const { finished } = require('node:stream');
const fs = require('node:fs');

const rs = fs.createReadStream('archive.tar');

finished(rs, (err) => {
  if (err) {
    console.error('Stream failed.', err);
  } else {
    console.log('Stream is done reading.');
  }
});

rs.resume(); // Drain the stream.
copy

在串流過早被銷毀（例如中止的 HTTP 請求）且不會發出 'end' 或 'finish' 的錯誤處理場景中特別有用。

finished API 提供 Promise 版本。

stream.finished() 在 callback 被叫用後會留下懸空的事件監聽器（特別是 'error'、'end'、'finish' 和 'close'）。原因是為了防止非預期的 'error' 事件（由於不正確的串流實作）導致非預期的崩潰。如果不希望這種行為，則需要在回呼中叫用回傳的清理函數。

const cleanup = finished(rs, (err) => {
  cleanup();
  // ...
});
copy