本頁面由 Cloud Translation API 翻譯而成。

訊息串：這份指南

瞭解如何透過 Streams API 使用可讀取、可寫入及轉換的串流。

Thomas Steiner

Streams API 可讓您透過程式輔助存取透過網路接收或以任何方式在本機建立的資料串流，並使用 JavaScript 加以處理。串流功能會將您要接收、傳送或轉換的資源分解為小型區塊，然後逐位處理這些區塊。雖然瀏覽器在接收 HTML 或影片等素材資源時，會在網頁上顯示串流內容，但在 2015 年推出 fetch 串流之前，JavaScript 從未提供這項功能。

注意： 嚴格來說，XMLHttpRequest 可以進行串流，但真棒。

先前，如果您想處理某種資源 (例如影片或文字檔等)，就必須下載整個檔案，然後等待系統將其反序列化為適當的格式，再進行處理。有了可供 JavaScript 使用的串流，這一切都會改變。您現在可以在用戶端取得 JavaScript 後，逐步使用 JavaScript 處理原始資料，不必產生緩衝區、字串或 blob。這項功能可用於多種用途，以下列舉其中幾項：

影片特效：透過即時套用特效的轉換串流，傳送可讀的影片串流。
資料 (解) 壓縮：透過轉換串流將檔案串流管道化，以便有選擇地對檔案進行 (解) 壓縮。
圖片解碼：透過轉換串流插入 HTTP 回應串流，這個串流會將位元組解碼為點陣圖資料，然後透過另一個轉換串流來將點陣圖轉換為 PNG。如果安裝在 Service Worker 的 fetch 處理常式中，您就能以透明化的方式填滿 AVIF 等新圖片格式。

瀏覽器支援

ReadableStream 和 WritableStream

瀏覽器支援

資料來源

TransformStream

瀏覽器支援

資料來源

核心概念

在詳細說明各種直播類型前，我先來介紹一些核心概念。

區塊

區塊是寫入或讀取串流的單一資料。可以是任何類型，甚至可以包含不同類型的區塊。在大多數情況下，一個區塊並非特定串流中最原子化的資料單位。舉例來說，位元組串流可能包含由 16 KiB Uint8Array 單位組成的區塊，而不是單一位元組。

可讀取的串流

可讀取的串流代表可讀取的資料來源。換句話說，資料會從可讀取的串流「傳出」。具體來說，可讀取的串流是 ReadableStream 類別的例項。

可寫入的串流

可寫入的串流代表可寫入資料的目的地。換句話說，資料會「進入」可寫入的串流。具體來說，可寫入的串流是 WritableStream 類別的例項。

轉換串流

轉換串流由一組串流組成：可寫串流 (稱為可寫端) 和可讀串流 (稱為可讀端)。現實世界的隱喻，就是「同步翻譯工具」，可即時將某種語言翻譯成另一種語言。在轉換串流的特定方式中，寫入可寫入端會導致新資料可供讀取端讀取。具體來說，任何含有 writable 屬性和 readable 屬性的物件都可以做為轉換串流。不過，標準 TransformStream 類別可讓您更輕鬆地建立正確交錯的這類組合。

管鏈

串流主要用於管道彼此之間。可讀取的串流可以使用可讀取串流的 pipeTo() 方法，直接傳送至可寫入的串流，也可以先使用可讀取串流的 pipeThrough() 方法，透過一或多個轉換串流傳送。以這種方式匯入一組管道，稱為管道鏈。

回壓

管道鏈結建構完成後，就會傳播信號，指出資料塊應以何種速度流經管道。如果鏈結中的任何步驟尚未接受區塊，就會透過管道鏈結向後傳播訊號，直到最終原始來源停止快速產生區塊為止。這項流量規範程序稱為回壓。

開球

可讀取的資料流可使用其 tee() 方法進行分流 (以大寫字母「T」的形狀命名)。這樣做會「鎖定」串流，也就是說，串流無法再直接使用，但會建立兩個新的串流，稱為分支版本，這些串流可獨立使用。分流也是重要的步驟，因為串流無法倒帶或重新開始，我們會在稍後進一步說明。

管道鏈結圖表，其中包含可讀取的串流，來自對 fetch API 的呼叫，然後透過轉換串流傳送，該串流的輸出內容會分流，並傳送至瀏覽器 (第一個結果可讀取串流) 和服務工作者快取 (第二個結果可讀取串流)。 — 管道鏈結。

可讀取串流的機制

可讀取的串流是資料來源，在 JavaScript 中以 ReadableStream 物件表示，該物件會從基礎來源流入。ReadableStream() 建構函式會透過指定的處理常式建立並傳回可讀取的串流物件。基礎來源分為兩種類型：

推送來源會在您存取時持續推送資料，您可以自行決定是否要開始、暫停或取消存取串流。例如即時影像串流、伺服器傳送的事件或 WebSocket。
連線至提取來源後，您必須明確要求資料。例如透過 fetch() 或 XMLHttpRequest 呼叫的 HTTP 作業。

串流資料會以稱為「區塊」的小片段依序讀取。放置在串流中的區塊稱為「已排入佇列」。這表示這些訊息正在佇列中等待讀取。內部佇列會追蹤尚未讀取的區塊。

佇列策略是一種物件，可決定串流應如何根據內部佇列的狀態，發出回壓信號。佇列策略會為每個區塊指派大小，並將佇列中所有區塊的總大小與指定數字 (稱為高水位標記) 進行比較。

串流中的區塊會由讀取器讀取。這個讀取器會一次擷取一小部分資料，讓您執行所需的任何作業。讀取器加上與之相關的其他處理程式碼，稱為「消費者」。

這個情境中的下一個結構體稱為「控制器」。每個可讀取的串流都會提供相關聯的控制器，顧名思義，您可以透過控制器控制串流。

一次只能讀取一個串流；當讀取者建立並開始讀取串流 (也就是成為「使用中讀取器」) 時，該串流就會「鎖定」。如果您想讓其他讀取器接管讀取串流，通常需要先釋放第一個讀取器，才能執行其他操作 (但您可以分流串流)。

建立可讀取的串流

您可以呼叫建構函式 ReadableStream() 來建立可讀取的串流。建構函式有選用引數 underlyingSource，可代表物件，其中包含定義建構的串流例項行為方式的方法和屬性。

`underlyingSource`

這可以使用下列選用的開發人員定義方法：

start(controller)：在物件建構時立即呼叫。該方法可存取串流來源，並執行設定串流功能所需的其他任何操作。如果這項程序是非同步完成，該方法可能會傳回承諾，表示成功或失敗。傳遞至此方法的 controller 參數為 ReadableStreamDefaultController。
pull(controller)：可用於控制串流，因為擷取更多區塊。只要串流的內部區塊佇列未滿，就會重複呼叫，直到佇列達到高水位為止。如果呼叫 pull() 的結果是承諾，則在承諾完成之前，系統不會再次呼叫 pull()。如果應許遭到拒絕，串流就會發生錯誤。
cancel(reason)：在串流消費者取消串流時呼叫。

const readableStream = new ReadableStream({
  start(controller) {
    /* … */
  },

  pull(controller) {
    /* … */
  },

  cancel(reason) {
    /* … */
  },
});

ReadableStreamDefaultController 支援下列方法：

ReadableStreamDefaultController.close() 會關閉相關聯的串流。
ReadableStreamDefaultController.enqueue() 會在相關聯的串流中將指定的區塊排入佇列。
ReadableStreamDefaultController.error() 會導致日後與相關串流的任何互動都會發生錯誤。

/* … */
start(controller) {
  controller.enqueue('The first chunk!');
},
/* … */

`queuingStrategy`

ReadableStream() 建構函式的第二個引數 (同樣為選用引數) 是 queuingStrategy。此物件可選擇性定義串流的排入佇列策略，該策略會採用兩個參數：

highWaterMark：非負數的數字，表示使用這個排入佇列的串流高度標示高度。
size(chunk)：函式會計算並傳回指定區塊值的有限非負大小。結果將用於判斷背壓，並透過適當的 ReadableStreamDefaultController.desiredSize 屬性顯示。也會控管呼叫基礎來源的 pull() 方法的時機。

const readableStream = new ReadableStream({
    /* … */
  },
  {
    highWaterMark: 10,
    size(chunk) {
      return chunk.length;
    },
  },
);

注意：您可以自行定義自訂 queuingStrategy，也可以為這個物件的值使用 ByteLengthQueuingStrategy 或 CountQueuingStrategy 的例項。如未提供 queuingStrategy，預設值會與含有 1 的 highWaterMark 的 CountQueuingStrategy 相同。

`getReader()` 和 `read()` 方法

如要從可讀取的串流讀取資料，您需要使用讀取器，也就是 ReadableStreamDefaultReader。ReadableStream 介面的 getReader() 方法會建立讀取器，並將串流鎖定在該讀取器上。串流鎖定後，必須等到這個串流發布後，才能收購其他讀者。

ReadableStreamDefaultReader 介面的 read() 方法會傳回承諾，提供對串流內部佇列中下一個區塊的存取權。它會根據串流的狀態，以結果滿足或拒絕。可能的情況如下：

如果有可用的區塊，承諾就會使用
{ value: chunk, done: false } 形式的物件來履行。
如果串流關閉，承諾會以
{ value: undefined, done: true } 表單的物件來履行。
如果串流發生錯誤，承諾會因相關錯誤而遭到拒絕。

const reader = readableStream.getReader();
while (true) {
  const { done, value } = await reader.read();
  if (done) {
    console.log('The stream is done.');
    break;
  }
  console.log('Just read a chunk:', value);
}

`locked` 屬性

您可以存取可讀取的串流的 ReadableStream.locked 屬性，檢查可讀取的串流是否遭到鎖定。

const locked = readableStream.locked;
console.log(`The stream is ${locked ? 'indeed' : 'not'} locked.`);

可讀取的串流程式碼範例

以下程式碼範例會顯示所有步驟的運作情形。您首先建立 ReadableStream，並在其 underlyingSource 引數 (即 TimestampSource 類別) 中定義 start() 方法。這個方法會告知串流的 controller 在 10 秒內每秒 enqueue() 一次時間戳記。最後，它會指示控制器 close() 串流。您可以透過 getReader() 方法建立讀取器，然後呼叫 read()，直到串流變為 done 為止，藉此使用此串流。

class TimestampSource {
  #interval

  start(controller) {
    this.#interval = setInterval(() => {
      const string = new Date().toLocaleTimeString();
      // Add the string to the stream.
      controller.enqueue(string);
      console.log(`Enqueued ${string}`);
    }, 1_000);

    setTimeout(() => {
      clearInterval(this.#interval);
      // Close the stream after 10s.
      controller.close();
    }, 10_000);
  }

  cancel() {
    // This is called if the reader cancels.
    clearInterval(this.#interval);
  }
}

const stream = new ReadableStream(new TimestampSource());

async function concatStringStream(stream) {
  let result = '';
  const reader = stream.getReader();
  while (true) {
    // The `read()` method returns a promise that
    // resolves when a value has been received.
    const { done, value } = await reader.read();
    // Result objects contain two properties:
    // `done`  - `true` if the stream has already given you all its data.
    // `value` - Some data. Always `undefined` when `done` is `true`.
    if (done) return result;
    result += value;
    console.log(`Read ${result.length} characters so far`);
    console.log(`Most recently read chunk: ${value}`);
  }
}
concatStringStream(stream).then((result) => console.log('Stream complete', result));

非同步疊代

在每次 read() 迴圈疊代時檢查串流是否為 done，可能不是最方便的 API。幸好在不久的將是更好的做法：非同步疊代。

for await (const chunk of stream) {
  console.log(chunk);
}

目前使用非同步疊代功能的解決方法，是使用 polyfill 實作行為。

if (!ReadableStream.prototype[Symbol.asyncIterator]) {
  ReadableStream.prototype[Symbol.asyncIterator] = async function* () {
    const reader = this.getReader();
    try {
      while (true) {
        const {done, value} = await reader.read();
        if (done) {
          return;
          }
        yield value;
      }
    }
    finally {
      reader.releaseLock();
    }
  }
}

分流可讀取的串流

ReadableStream 介面的 tee() 方法會擷取目前的可讀取串流，傳回雙元素陣列，其中包含兩個產生的分支版本，做為新的 ReadableStream 執行個體。這樣一來，兩個讀取器就能同時讀取串流。舉例來說，如果您想從伺服器擷取回應並串流到瀏覽器，但同時也想串流至服務工作站快取，就可以在 Service Worker 中執行此操作。由於回應主體無法重複使用，因此您需要兩個副本才能執行這項操作。如要取消串流，您必須取消兩個產生的分支。分流串流通常會在整個過程中鎖定，防止其他讀取器鎖定。

const readableStream = new ReadableStream({
  start(controller) {
    // Called by constructor.
    console.log('[start]');
    controller.enqueue('a');
    controller.enqueue('b');
    controller.enqueue('c');
  },
  pull(controller) {
    // Called `read()` when the controller's queue is empty.
    console.log('[pull]');
    controller.enqueue('d');
    controller.close();
  },
  cancel(reason) {
    // Called when the stream is canceled.
    console.log('[cancel]', reason);
  },
});

// Create two `ReadableStream`s.
const [streamA, streamB] = readableStream.tee();

// Read streamA iteratively one by one. Typically, you
// would not do it this way, but you certainly can.
const readerA = streamA.getReader();
console.log('[A]', await readerA.read()); //=> {value: "a", done: false}
console.log('[A]', await readerA.read()); //=> {value: "b", done: false}
console.log('[A]', await readerA.read()); //=> {value: "c", done: false}
console.log('[A]', await readerA.read()); //=> {value: "d", done: false}
console.log('[A]', await readerA.read()); //=> {value: undefined, done: true}

// Read streamB in a loop. This is the more common way
// to read data from the stream.
const readerB = streamB.getReader();
while (true) {
  const result = await readerB.read();
  if (result.done) break;
  console.log('[B]', result);
}

可讀取的位元組串流

針對代表位元的串流，我們會提供可讀取串流的擴充版本，以便有效處理位元，特別是盡量減少複製次數。位元串流可讓您取得自備緩衝區 (BYOB) 讀取器。在 WebSocket 的情況下，預設實作可以提供範圍不同的輸出內容，例如字串或陣列緩衝區，而位元組串流則保證會輸出位元組。此外，自備讀取器具有穩定性優勢。這是因為如果緩衝區分離，可以確保不會寫入相同的緩衝區兩次，進而避免競爭狀態。BYOB 讀取器可重複使用緩衝區，因此可減少瀏覽器執行垃圾收集所需的次數。

建立可讀取的位元組串流

只要將額外的 type 參數傳遞至 ReadableStream() 建構函式，即可建立可讀取的位元組資料流。

new ReadableStream({ type: 'bytes' });

`underlyingSource`

可讀取位元組串流的基礎來源會給予 ReadableByteStreamController 以操控。其 ReadableByteStreamController.enqueue() 方法會採用 chunk 引數，其值為 ArrayBufferView。屬性 ReadableByteStreamController.byobRequest 會傳回目前的 BYOB 提取要求，如果沒有則傳回 null。最後，ReadableByteStreamController.desiredSize 屬性會傳回所需大小，以填滿受控制串流的內部佇列。

`queuingStrategy`

ReadableStream() 建構函式的第二個引數 (同樣為選用引數) 是 queuingStrategy。這是一個物件，可選擇定義串流的佇列策略，並採用一個參數：

highWaterMark：非負數的位元組數量，表示使用這個排佇列策略的串流高浮水印。可用來判斷背壓，並透過適當的 ReadableByteStreamController.desiredSize 屬性顯示。它也會決定何時呼叫基礎來源的 pull() 方法。

注意：與其他串流類型的排隊策略不同，可讀取位元組串流的排隊策略沒有 size(chunk) 函式。每個區塊的大小一律由其 byteLength 屬性決定。

注意：如果未提供 queuingStrategy，系統會使用 highWaterMark 為 0 的預設值。

`getReader()` 和 `read()` 方法

接著，您可以據此設定 mode 參數，以便存取 ReadableStreamBYOBReader：ReadableStream.getReader({ mode: "byob" })。這樣一來，您就能更精確地控管緩衝區配置，避免複製。如要讀取位元組資料流，您必須呼叫 ReadableStreamBYOBReader.read(view)，其中 view 是 ArrayBufferView。

可讀取的位元組串流程式碼範例

const reader = readableStream.getReader({ mode: "byob" });

let startingAB = new ArrayBuffer(1_024);
const buffer = await readInto(startingAB);
console.log("The first 1024 bytes, or less:", buffer);

async function readInto(buffer) {
  let offset = 0;

  while (offset < buffer.byteLength) {
    const { value: view, done } =
        await reader.read(new Uint8Array(buffer, offset, buffer.byteLength - offset));
    buffer = view.buffer;
    if (done) {
      break;
    }
    offset += view.byteLength;
  }

  return buffer;
}

下列函式會傳回可讀取的位元組串流，可讓您有效率地讀取隨機產生的陣列，而無需複製資料。而不是使用 1,024 的預定區塊大小，而是嘗試填充開發人員提供的緩衝區，以便完整控制。

const DEFAULT_CHUNK_SIZE = 1_024;

function makeReadableByteStream() {
  return new ReadableStream({
    type: 'bytes',

    pull(controller) {
      // Even when the consumer is using the default reader,
      // the auto-allocation feature allocates a buffer and
      // passes it to us via `byobRequest`.
      const view = controller.byobRequest.view;
      view = crypto.getRandomValues(view);
      controller.byobRequest.respond(view.byteLength);
    },

    autoAllocateChunkSize: DEFAULT_CHUNK_SIZE,
  });
}

可寫入串流的機制

可寫入的串流是可寫入資料的目的地，在 JavaScript 中以 WritableStream 物件表示。這可做為基礎接收端 (較低層級的 I/O 接收端，用於寫入原始資料) 頂端的抽象層。

資料會透過writer寫入串流，每次寫入一個區塊。就像讀取器中的區塊一樣，區塊可以採用多種形式。您可以使用任何程式碼產生可供寫入的區塊；寫入器加上相關程式碼稱為產生器。

當寫入器建立並開始寫入資料流 (有效寫入器) 時，就會被視為鎖定。一次只能有一個寫入器寫入可寫入式串流。如果想讓其他寫入者開始寫入您的串流，通常需要先釋出該串流，然後再附加其他寫入者。

內部佇列會追蹤已寫入串流但尚未由基礎接收端處理的區塊。

排隊策略是一種物件，可根據內部佇列的狀態，決定串流應如何傳送回壓信號。佇列策略會為每個區塊指派大小，並將佇列中所有區塊的總大小與指定數字 (稱為高水位標記) 進行比較。

最終的建構項目稱為控制器。每個可寫入的串流都有相關聯的控制器，可讓您控制串流 (例如中止串流)。

建立可寫入的串流

Streams API 的 WritableStream 介面提供標準的抽象概念，可將串流資料寫入目的地 (稱為接收器)。這個物件內建回壓和排隊功能。如要建立可寫入的串流，您可以呼叫其建構函式 WritableStream()。其中包含選用的 underlyingSink 參數，用於代表物件，方法和屬性可定義建構的串流執行個體行為。

`underlyingSink`

underlyingSink 可包含下列選用的開發人員定義方法。傳遞至部分方法的 controller 參數為 WritableStreamDefaultController。

start(controller)：在建構物件時會立即呼叫此方法。此方法的內容應以取得對基礎接收器的存取權為目標。如果要以非同步方式執行這項程序，則可傳回承諾，以便指出成功或失敗。
write(chunk, controller)：當新的資料區塊 (在 chunk 參數中指定) 準備寫入基礎接收來源時，系統會呼叫這個方法。它可以傳回承諾，用來表示寫入作業的成功或失敗。只有在先前的寫入成功後，系統才會呼叫這個方法，串流關閉或取消後也絕不會呼叫此方法。
close(controller)：如果應用程式傳送信號表示已完成將區塊寫入串流，系統就會呼叫這個方法。內容應進行任何必要的作業，才能完成對基礎接收器的寫入作業，以及釋出其存取權。如果這個程序是異步的，則可以傳回承諾，用來表示成功或失敗。只有在所有排隊的寫入作業成功後，系統才會呼叫這個方法。
abort(reason)：如果應用程式傳送訊號表示要突然關閉串流並將其置於錯誤狀態，系統就會呼叫這個方法。它可以清理任何已保留的資源，就像 close() 一樣，但即使寫入作業已排入佇列，系統也會呼叫 abort()。系統會丟棄這些區塊。如果是非同步的程序，可能會傳回承諾，表示成功或失敗。reason 參數包含 DOMString，說明流程中斷的原因。

const writableStream = new WritableStream({
  start(controller) {
    /* … */
  },

  write(chunk, controller) {
    /* … */
  },

  close(controller) {
    /* … */
  },

  abort(reason) {
    /* … */
  },
});

Streams API 的 WritableStreamDefaultController 介面代表控制器，可在設定期間 (更多區塊提交供寫入) 或寫入結束時，控管 WritableStream 的狀態。建構 WritableStream 時，基礎接收端會獲得對應的 WritableStreamDefaultController 例項來操作。WritableStreamDefaultController 只有一個方法：WritableStreamDefaultController.error()，這會導致日後與相關串流的任何互動都會發生錯誤。WritableStreamDefaultController 也支援 signal 屬性，可傳回 AbortSignal 的例項，讓您在需要時停止 WritableStream 作業。

/* … */
write(chunk, controller) {
  try {
    // Try to do something dangerous with `chunk`.
  } catch (error) {
    controller.error(error.message);
  }
},
/* … */

`queuingStrategy`

WritableStream() 建構函式的第二個引數 (同樣為選用引數) 是 queuingStrategy。這是一個物件，可選擇定義串流的佇列策略，該策略會採用兩個參數：

highWaterMark：非負數的數字，表示使用這個排入佇列的串流高度標示高度。
size(chunk)：函式會計算並傳回指定區塊值的有限非負大小。結果會用來判斷回壓，並透過適當的 WritableStreamDefaultWriter.desiredSize 屬性顯示。

注意：您可以自行定義自訂 queuingStrategy，也可以為這個物件值使用 ByteLengthQueuingStrategy 或 CountQueuingStrategy 的例項。如未提供 queuingStrategy，預設值會與含有 1 的 highWaterMark 的 CountQueuingStrategy 相同。

`getWriter()` 和 `write()` 方法

如要寫入可寫入的串流，您需要寫入者，也就是 WritableStreamDefaultWriter。WritableStream 介面的 getWriter() 方法會傳回新的 WritableStreamDefaultWriter 例項，並將串流鎖定至該例項。當串流遭到鎖定時，除非釋放目前的寫入器，否則無法取得其他寫入器。

WritableStreamDefaultWriter 介面的 write() 方法會將傳遞的資料區塊寫入 WritableStream 及其基礎接收器，然後傳回承諾，表示寫入作業成功或失敗。請注意，「成功」的意義取決於基礎接收器；它可能表示區塊已接受，但不一定能安全儲存至最終目的地。

const writer = writableStream.getWriter();
const resultPromise = writer.write('The first chunk!');

`locked` 屬性

您可以存取可寫串流的 WritableStream.locked 屬性，確認串流是否已上鎖。

const locked = writableStream.locked;
console.log(`The stream is ${locked ? 'indeed' : 'not'} locked.`);

可寫入串流程式碼範例

以下程式碼範例顯示了所有操作步驟。

const writableStream = new WritableStream({
  start(controller) {
    console.log('[start]');
  },
  async write(chunk, controller) {
    console.log('[write]', chunk);
    // Wait for next write.
    await new Promise((resolve) => setTimeout(() => {
      document.body.textContent += chunk;
      resolve();
    }, 1_000));
  },
  close(controller) {
    console.log('[close]');
  },
  abort(reason) {
    console.log('[abort]', reason);
  },
});

const writer = writableStream.getWriter();
const start = Date.now();
for (const char of 'abcdefghijklmnopqrstuvwxyz') {
  // Wait to add to the write queue.
  await writer.ready;
  console.log('[ready]', Date.now() - start, 'ms');
  // The Promise is resolved after the write finishes.
  writer.write(char);
}
await writer.close();

將可讀取的串流管道傳送至可寫入的串流

可讀取的串流可透過可讀取串流的 pipeTo() 方法，管道傳送至可寫入的串流。ReadableStream.pipeTo() 會將目前的 ReadableStream 管道傳送至指定的 WritableStream，並傳回承諾，如果管道程序順利完成，則會履行承諾，如果發生任何錯誤，則會拒絕。

const readableStream = new ReadableStream({
  start(controller) {
    // Called by constructor.
    console.log('[start readable]');
    controller.enqueue('a');
    controller.enqueue('b');
    controller.enqueue('c');
  },
  pull(controller) {
    // Called when controller's queue is empty.
    console.log('[pull]');
    controller.enqueue('d');
    controller.close();
  },
  cancel(reason) {
    // Called when the stream is canceled.
    console.log('[cancel]', reason);
  },
});

const writableStream = new WritableStream({
  start(controller) {
    // Called by constructor
    console.log('[start writable]');
  },
  async write(chunk, controller) {
    // Called upon writer.write()
    console.log('[write]', chunk);
    // Wait for next write.
    await new Promise((resolve) => setTimeout(() => {
      document.body.textContent += chunk;
      resolve();
    }, 1_000));
  },
  close(controller) {
    console.log('[close]');
  },
  abort(reason) {
    console.log('[abort]', reason);
  },
});

await readableStream.pipeTo(writableStream);
console.log('[finished]');

建立轉換串流

Streams API 的 TransformStream 介面代表一組可轉換的資料。您可以呼叫轉換串流建構函式 TransformStream() 來建立轉換串流，該函式會從指定的處理常建立並傳回轉換串流物件。TransformStream() 建構函式會接受選用的 JavaScript 物件做為第一個引數，代表 transformer。這類物件可包含下列任一方法：

`transformer`

start(controller)：這個方法會在物件建構時立即呼叫。通常會使用 controller.enqueue() 將前置字串分割成佇列。這些區塊會從可讀端讀取，但不依賴對可寫端的任何寫入作業。如果這個初始程序是異步的，例如因為取得前置字串區塊需要花費一些力氣，函式可以傳回承諾，用來表示成功或失敗；遭到拒絕的承諾會導致串流發生錯誤。任何擲回的例外狀況都將由 TransformStream() 建構函式重新擲回。
transform(chunk, controller)：當原本寫入可寫入端的新區塊已準備好進行轉換時，系統會呼叫這個方法。串流實作可確保只有在先前的轉換成功完成後，才會呼叫這個函式，也絕不會在 start() 完成或呼叫 flush() 之後呼叫此函式。此函式會執行轉換串流的實際轉換作業。它可以使用 controller.enqueue() 將結果排入佇列。這可讓寫入可寫端的單一區塊，在可讀端產生零或多個區塊，具體取決於 controller.enqueue() 的呼叫次數。如果轉換程序是非同步的，這個函式可以傳回承諾，用來表示轉換的成功或失敗。遭到拒絕的應許會在轉換串流的可讀和可寫端都發生錯誤。如果未提供 transform() 方法，系統會使用身分轉換，將未變更的區塊從可寫端排入讀取端。
flush(controller)：在成功傳送至可寫入端的所有區塊經過 transform() 轉換後，且即將關閉可寫入端時，系統會呼叫這個方法。通常用於在可讀取的端將後置字串區塊排入佇列中，在最後關閉前使用該區塊。如果清除程序是非同步的，函式可以傳回承諾，以示成功或失敗；結果會傳達給 stream.writable.write() 的呼叫端。此外，遭到拒絕的承諾會在可讀取和可寫入的串流端都發生錯誤。擲回例外狀況等同於傳回遭拒的承諾。

const transformStream = new TransformStream({
  start(controller) {
    /* … */
  },

  transform(chunk, controller) {
    /* … */
  },

  flush(controller) {
    /* … */
  },
});

`writableStrategy` 和 `readableStrategy` 排隊策略

TransformStream() 建構函式的第二和第三個選用參數為選用的 writableStrategy 和 readableStrategy 佇列策略。分別在「可讀取」和「可寫入」串流部分中定義。

轉換串流程式碼範例

以下程式碼範例顯示簡單的轉換串流運作情形。

// Note that `TextEncoderStream` and `TextDecoderStream` exist now.
// This example shows how you would have done it before.
const textEncoderStream = new TransformStream({
  transform(chunk, controller) {
    console.log('[transform]', chunk);
    controller.enqueue(new TextEncoder().encode(chunk));
  },
  flush(controller) {
    console.log('[flush]');
    controller.terminate();
  },
});

(async () => {
  const readStream = textEncoderStream.readable;
  const writeStream = textEncoderStream.writable;

  const writer = writeStream.getWriter();
  for (const char of 'abc') {
    writer.write(char);
  }
  writer.close();

  const reader = readStream.getReader();
  for (let result = await reader.read(); !result.done; result = await reader.read()) {
    console.log('[value]', result.value);
  }
})();

透過轉換串流管道傳送可讀取的串流

ReadableStream 介面的 pipeThrough() 方法提供可連結的方式，可透過轉換串流或任何其他可寫入/可讀取的組合，管道傳送目前的串流。管道化串流通常會在管道期間鎖定串流，防止其他讀取器鎖定串流。

const transformStream = new TransformStream({
  transform(chunk, controller) {
    console.log('[transform]', chunk);
    controller.enqueue(new TextEncoder().encode(chunk));
  },
  flush(controller) {
    console.log('[flush]');
    controller.terminate();
  },
});

const readableStream = new ReadableStream({
  start(controller) {
    // called by constructor
    console.log('[start]');
    controller.enqueue('a');
    controller.enqueue('b');
    controller.enqueue('c');
  },
  pull(controller) {
    // called read when controller's queue is empty
    console.log('[pull]');
    controller.enqueue('d');
    controller.close(); // or controller.error();
  },
  cancel(reason) {
    // called when rs.cancel(reason)
    console.log('[cancel]', reason);
  },
});

(async () => {
  const reader = readableStream.pipeThrough(transformStream).getReader();
  for (let result = await reader.read(); !result.done; result = await reader.read()) {
    console.log('[value]', result.value);
  }
})();

下列程式碼範例 (稍微複雜) 說明如何實作「大聲」版本的 fetch()，這個版本會將傳回的回應承諾視為流量，並逐個區塊將所有文字轉為大寫。這種方法的好處是，您不必等待整份文件下載完畢，這在處理大型檔案時非常有用。

function upperCaseStream() {
  return new TransformStream({
    transform(chunk, controller) {
      controller.enqueue(chunk.toUpperCase());
    },
  });
}

function appendToDOMStream(el) {
  return new WritableStream({
    write(chunk) {
      el.append(chunk);
    }
  });
}

fetch('./lorem-ipsum.txt').then((response) =>
  response.body
    .pipeThrough(new TextDecoderStream())
    .pipeThrough(upperCaseStream())
    .pipeTo(appendToDOMStream(document.body))
);

示範

以下示範說明可讀、可寫和轉換串流的實際運作情形。其中也包含 pipeThrough() 和 pipeTo() 管道鏈結的範例，並且示範 tee()。您可以選擇在專屬視窗中執行示範，或查看原始碼。

瀏覽器提供的實用串流

瀏覽器內建了許多實用的串流。您可以輕鬆從 blob 建立 ReadableStream。Blob 介面的 stream() 方法會傳回 ReadableStream，該方法在讀取時會傳回 Blob 內含的資料。另外，請注意，File 物件是 Blob 的特定類型，可用於 Blob 可用的任何情境。

const readableStream = new Blob(['hello world'], { type: 'text/plain' }).stream();

TextDecoder.decode() 和 TextEncoder.encode() 的串流變體分別稱為 TextDecoderStream 和 TextEncoderStream。

const response = await fetch('https://streams.spec.whatwg.org/');
const decodedStream = response.body.pipeThrough(new TextDecoderStream());

您可以分別使用 CompressionStream 和 DecompressionStream 轉換串流，輕鬆壓縮或解壓縮檔案。下列程式碼範例說明如何下載 Streams 規格、直接在瀏覽器中壓縮 (gzip) 檔案，並將壓縮檔案直接寫入磁碟。

const response = await fetch('https://streams.spec.whatwg.org/');
const readableStream = response.body;
const compressedStream = readableStream.pipeThrough(new CompressionStream('gzip'));

const fileHandle = await showSaveFilePicker();
const writableStream = await fileHandle.createWritable();
compressedStream.pipeTo(writableStream);

File System Access API 的 FileSystemWritableFileStream 和實驗性的 fetch() 要求串流，就是實際可寫入串流的例子。

Serial API 會大量使用可讀取和可寫入的串流。

// Prompt user to select any serial port.
const port = await navigator.serial.requestPort();
// Wait for the serial port to open.
await port.open({ baudRate: 9_600 });
const reader = port.readable.getReader();

// Listen to data coming from the serial device.
while (true) {
  const { value, done } = await reader.read();
  if (done) {
    // Allow the serial port to be closed later.
    reader.releaseLock();
    break;
  }
  // value is a Uint8Array.
  console.log(value);
}

// Write to the serial port.
const writer = port.writable.getWriter();
const data = new Uint8Array([104, 101, 108, 108, 111]); // hello
await writer.write(data);
// Allow the serial port to be closed later.
writer.releaseLock();

最後，WebSocketStream API 會將串流與 WebSocket API 整合。

const wss = new WebSocketStream(WSS_URL);
const { readable, writable } = await wss.connection;
const reader = readable.getReader();
const writer = writable.getWriter();

while (true) {
  const { value, done } = await reader.read();
  if (done) {
    break;
  }
  const result = await process(value);
  await writer.write(result);
}

實用資源

特別銘謝

本文由 Jake Archibald、François Beaufort、Sam Dutton、Mattias Buelens、Surma、Joe Medley 和 Adam Rice 共同審查。Jake Archibald 的網誌文章對我瞭解串流非常有幫助。部分程式碼範例的靈感來自 GitHub 使用者 @bellbind 的探索，而部分文字則大量參考 MDN Web Streams 說明文件。Streams Standard 的作者在撰寫此規格時付出極大努力。主圖由 Ryan Lara 在 Unsplash 上提供。

訊息串：這份指南

瀏覽器支援

ReadableStream 和 WritableStream

TransformStream

核心概念

區塊

可讀取的串流

可寫入的串流

轉換串流

管鏈

回壓

開球

可讀取串流的機制

建立可讀取的串流

underlyingSource

queuingStrategy

getReader() 和 read() 方法

locked 屬性

可讀取的串流程式碼範例

非同步疊代

分流可讀取的串流

可讀取的位元組串流

建立可讀取的位元組串流

underlyingSource

queuingStrategy

getReader() 和 read() 方法

可讀取的位元組串流程式碼範例

可寫入串流的機制

建立可寫入的串流

underlyingSink

queuingStrategy

getWriter() 和 write() 方法

locked 屬性

可寫入串流程式碼範例

將可讀取的串流管道傳送至可寫入的串流

建立轉換串流

transformer

writableStrategy 和 readableStrategy 排隊策略

轉換串流程式碼範例

透過轉換串流管道傳送可讀取的串流

示範

瀏覽器提供的實用串流

實用資源

特別銘謝

`underlyingSource`

`queuingStrategy`

`getReader()` 和 `read()` 方法

`locked` 屬性

`underlyingSource`

`queuingStrategy`

`getReader()` 和 `read()` 方法

`underlyingSink`

`queuingStrategy`

`getWriter()` 和 `write()` 方法

`locked` 屬性

`transformer`

`writableStrategy` 和 `readableStrategy` 排隊策略