Node.js v25.0.0 文件

事件：'close'#

當流及其任何底層資源（例如檔案描述符）被關閉時，會發出 'close' 事件。該事件表明不會再發出更多事件，也不會發生進一步的計算。

如果一個可寫流是使用 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() 方法後，並且所有資料都已重新整理到底層系統時，會發出 '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> 從此可寫流取消管道傳輸的源流

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

當一個可讀流管道傳輸到此可寫流時，如果該可寫流發出錯誤，也會發出此事件。

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() 方法時，緩衝的資料將被重新整理。

writable.cork() 的主要目的是為了適應一種情況，即在短時間內向流中寫入多個小資料塊。writable.cork() 不會立即將它們轉發到底層目標，而是緩衝所有資料塊，直到呼叫 writable.uncork()，屆時如果存在 writable._writev()，它會將所有資料塊傳遞給該方法。這可以防止因等待處理第一個小資料塊而導致資料被緩衝的“隊頭阻塞”情況。但是，在沒有實現 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() 方法表示不會再有資料寫入到可寫流。可選的 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() 方法為可寫流設定預設的 encoding。

writable.uncork()#

writable.uncork() 方法會重新整理自 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() 以完全解除流的阻塞的次數。

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' 事件。

當流未排空時，對 write() 的呼叫將緩衝 chunk 並返回 false。一旦所有當前緩衝的塊都被排空（即被作業系統接受以進行傳遞），將會觸發 'drain' 事件。一旦 write() 返回 false，在 'drain' 事件觸發前，請勿寫入更多的資料塊。雖然在流未排空時呼叫 write() 是允許的，但 Node.js 會緩衝所有寫入的塊，直到達到最大記憶體使用量，此時它將無條件地中止。即使在中止之前，高記憶體使用也會導致垃圾回收器效能下降和高 RSS（常駐記憶體大小，即使記憶體不再需要，通常也不會釋放回系統）。由於如果遠端對等方不讀取資料，TCP 套接字可能永遠不會排空，因此向一個未排空的套接字寫入資料可能導致可被遠端利用的漏洞。

在流未排空時寫入資料對於 Transform 流尤其成問題，因為 Transform 流預設是暫停的，直到它們被管道連線（pipe）或添加了 'data' 或 'readable' 事件處理程式。

如果要寫入的資料可以按需生成或獲取，建議將邏輯封裝到 Readable 流中並使用 stream.pipe()。但是，如果傾向於呼叫 write()，則可以透過使用 'drain' 事件來遵循背壓機制並避免記憶體問題。

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

處於物件模式（object mode）的 Writable 流將始終忽略 encoding 引數。

事件：'close'#

當流及其任何底層資源（例如檔案描述符）被關閉時，會發出 'close' 事件。該事件表明不會再發出更多事件，也不會發生進一步的計算。

如果 Readable 流建立時帶有 emitClose 選項，它將總是觸發 'close' 事件。

事件：'data'#

• chunk <Buffer> | <string> | <any> 資料塊。對於非物件模式的流，chunk 將是字串或 Buffer。對於物件模式的流，chunk 可以是除 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>

'error' 事件可以由 Readable 實現隨時觸發。通常，這可能發生在底層流因內部故障而無法生成資料，或者當流實現嘗試推送一個無效的資料塊時。

監聽器回撥函式將接收到一個單獨的 Error 物件。

事件：'pause'#

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

事件：'readable'#

當有資料可從流中讀取時，會觸發 'readable' 事件，可讀取的資料量最多達到配置的高水位線（state.highWaterMark）。實際上，它表示流的緩衝區中有新的資訊。如果此緩衝區中有可用資料，可以呼叫 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' 在控制流方面具有優先權，即只有在呼叫 stream.read() 時才會觸發 'data'。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> 目標流，如果它是 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

處於物件模式的 Readable 流在呼叫 readable.read(size) 時將始終返回單個專案，無論 size 引數的值是多少。

如果 readable.read() 方法返回一個數據塊，那麼也會觸發一個 'data' 事件。

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

readable.readable#

• 型別：<boolean>

如果可以安全地呼叫 readable.read()，則為 true，這意味著流尚未被銷燬或觸發 'error' 或 'end'。

readable.readableAborted#

• 型別：<boolean>

返回流是否在觸發 'end' 之前被銷燬或出錯。

readable.readableDidRead#

• 型別：<boolean>

返回 'data' 事件是否已被觸發。

readable.readableEncoding#

• 型別：<null> | <string>

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

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() 方法將一塊資料推回到內部緩衝區。這在某些情況下很有用，例如當一個流被需要“反消費”（un-consume）一些它已從源中樂觀拉取的資料的程式碼消費時，以便這些資料可以傳遞給其他方。

在 '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。（有關更多資訊，請參見與舊版 Node.js 的相容性。）

當使用一個會觸發 '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() 並返回一個在流完成時履行的 promise。

readable.compose(stream[, options])#

• stream <Stream> | <Iterable> | <AsyncIterable> | <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(['this is', 'compose as operator']).compose(splitToWords);
const words = await wordsStream.toArray();

console.log(words); // prints ['this', 'is', 'compose', 'as', 'operator'] copy

參見 stream.compose 瞭解更多資訊。

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> 一個用於對映流中每個資料塊的函式。
  • 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 value），該資料塊將被傳遞到結果流。如果 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]

// 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，直到等待的返回值為 true（或任何真值）。一旦對某個資料塊的 fn 呼叫等待的返回值為真值，流就會被銷燬，並且 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 呼叫等待的返回值為真值，流就會被銷燬，並且 promise 會以該資料塊履行。如果所有對資料塊的 fn 呼叫都返回假值（falsy value），則 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，以檢查所有等待的返回值是否都為真值。一旦對某個資料塊的 fn 呼叫等待的返回值為假值，流就會被銷燬，並且 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 進行扁平化對映的流。

此方法透過將給定的回撥應用於流的每個資料塊然後將結果扁平化來返回一個新流。

可以從 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> 一個在流的每個資料塊上呼叫的歸納函式。
  • previous <any> 上一次呼叫 fn 獲得的值，或者如果指定了 initial 值則是該值，否則是流的第一個資料塊。
  • data <any> 來自流的一個數據塊。
  • options <Object>
    • signal <AbortSignal> 如果流被銷燬，該訊號會中止，從而允許儘早中止 fn 呼叫。
• initial <any> 在歸納中使用的初始值。
• options <Object>
  • signal <AbortSignal> 允許在訊號中止時銷燬流。
• 返回: <Promise> 一個歸納最終值的 promise。

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

如果沒有提供 initial 值，則使用流的第一個資料塊作為初始值。如果流為空，則 promise 會被拒絕，並帶有一個具有 ERR_INVALID_ARGS 程式碼屬性的 TypeError。

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

duplex.allowHalfOpen#

• 型別：<boolean>

如果為 false，那麼當可讀端結束時，流將自動結束可寫端。最初由 allowHalfOpen 建構函式選項設定，預設為 true。

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

transform.destroy([error])#

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

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

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