编码母题 / P1 / Node 与 JavaScript / 04

支持背压的流式数据管道

先读清题意与边界，独立尝试，再展开思路并对照可运行的参考代码。

重要程度高价值难度困难预计完成45–60 分钟

难度按 LeetCode 简单／中等／困难三档估计；工程题并非 LeetCode 官方题目。用时指独立编码并检查主要边界的参考时间，因经验而异。

题目是什么

把异步可迭代源持续写入 Node Writable。write 返回 false 时暂停生产者等待 drain；还要处理 error、取消和正常 finish。

输入输出示例

写入触发背压后，只有 drain 到达才读取并写入下一块。

练习起手式与验证

复制代码后补全空方法，再运行下方三组验证；复杂题所需的辅助类型与输入准备一并放在代码块中。

/*
 * 题目：支持背压的流式数据管道
 * 要求：把异步可迭代源持续写入 Node Writable。write 返回 false
 * 时暂停生产者等待 drain；还要处理 error、取消和正常 finish。
 * 示例：写入触发背压后，只有 drain 到达才读取并写入下一块。
 */
const { Writable } = require("node:stream");

function waitEvent(stream, successEvent, signal) {}
async function writeWithBackpressure(source, writable, { signal } = {}) {}

(async () => {
  const chunks = [];
  let writes = 0;
  const writable = new Writable({
    highWaterMark: 1,
    write(chunk, _encoding, callback) {
      writes += 1;
      setTimeout(() => {
        chunks.push(chunk.toString());
        callback();
      }, 2);
    },
  });
  async function* source() {
    yield "a";
    yield "b";
    yield "c";
  }
  await writeWithBackpressure(source(), writable);
  console.log(writes); // 预期：3
  console.log(chunks); // 预期：["a", "b", "c"]
  console.log(writable.writableFinished); // 预期：true
})().catch(console.error);

解题思路

展开思路、不变量与常见边界

关键切入点：for await 消费源，遇到 write=false 等待 drain/error/abort 之一，并在所有分支清理监听器。

核心模型：for await 消费源；write false 后 await drain/error/abort 三者之一；正常结束调用 end 并等待 finish。监听必须在任一分支完成后清理。不变量：收到 drain 前不再 write；失败/取消后不继续消费；函数 resolve 时输出已 finish。迁移：文件、HTTP 响应、SSE、模型 Token 和批量数据导出。

参考代码与验证

“关键机制”注释用于解释实现，下载链接提供未经页面注释修改的原始 JS 文件。

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

// 关键机制：for await 消费源，遇到 write=false 等待 drain/error/abort 之一，并在所有分支清理监听器。
function waitEvent(stream, successEvent, signal) {
  if (signal?.aborted)
    return Promise.reject(signal.reason || new Error("aborted"));
  return new Promise((resolve, reject) => {
    const onSuccess = () => {
      cleanup();
      resolve();
    };
    const onError = (e) => {
      cleanup();
      reject(e);
    };
    const onAbort = () => {
      cleanup();
      reject(signal.reason || new Error("aborted"));
    };
    function cleanup() {
      stream.off(successEvent, onSuccess);
      stream.off("error", onError);
      signal?.removeEventListener("abort", onAbort);
    }
    stream.once(successEvent, onSuccess);
    stream.once("error", onError);
    signal?.addEventListener("abort", onAbort, { once: true });
  });
}
async function writeWithBackpressure(source, writable, { signal } = {}) {
  try {
    for await (const chunk of source) {
      if (signal?.aborted) throw signal.reason || new Error("aborted");
      if (!writable.write(chunk)) await waitEvent(writable, "drain", signal);
    }
    const finished = waitEvent(writable, "finish", signal);
    writable.end();
    await finished;
  } catch (error) {
    if (!writable.destroyed) writable.destroy();
    throw error;
  }
}

// 三组以上可直接运行的验证；异步题使用独立执行块。
(async () => {
  const chunks = [];
  let writes = 0;
  const writable = new Writable({
    highWaterMark: 1,
    write(chunk, _encoding, callback) {
      writes += 1;
      setTimeout(() => {
        chunks.push(chunk.toString());
        callback();
      }, 2);
    },
  });
  async function* source() {
    yield "a";
    yield "b";
    yield "c";
  }
  await writeWithBackpressure(source(), writable);
  console.log(writes); // 预期：3
  console.log(chunks); // 预期：["a", "b", "c"]
  console.log(writable.writableFinished); // 预期：true


})().catch(console.error);

复杂度分析

时间复杂度：O(n)，不计底层 I/O。从源读取并向目标写入 n 个数据块各一次。

空间复杂度：O(1) 辅助空间。只等待当前写入和 drain；流自身的缓冲空间另计。

下载原始 JS 文件