ReadableStream 怎样用 TransformStream 转换数据

我第一次把流式响应改成“边到边处理”时，页面偶尔会把一条消息拆成两条，或者把两条消息拼成一条。问题不在 TransformStream 不能转换数据，而是我把底层传来的 chunk 误当成了完整业务消息。ReadableStream.pipeThrough() 负责连接流，真正的消息边界仍然需要由转换逻辑自己维护。

官方地址：https://developer.mozilla.org/en-US/docs/Web/API/ReadableStream/pipeThrough

readable.pipeThrough(transformStream) 会把源流写入转换流的 writable 端，并返回转换流的 readable 端。最小用法是实现 transform(chunk, controller)，但网络文本还要处理解码和跨 chunk 的业务边界。

下面只围绕 pipeThrough 展开：先看三段关系，再写一个最小转换器，最后把它放进网络文本管道，并处理关闭、异常和取消。文中的图片是静态结构说明图，不是浏览器截图，也不是实际运行证据。

先把 pipeThrough 的三段关系看清

TransformStream 不是一个单独的“转换函数”，而是一对彼此连接的流：writable 接受输入，readable 提供转换后的输出。pipeThrough() 做的事情可以概括为：把当前 ReadableStream 接到转换器的输入端，然后把转换器的输出端返回给调用方。

因此，下面两种写法的含义不同：

// 返回转换后的 ReadableStream，可以继续接下一个转换器。
const transformed = source.pipeThrough(transform);

// pipeTo 才是把转换后的流送到最终 WritableStream 的终点操作。
await transformed.pipeTo(destination);

如果把 source 同时交给多个读取者，或者在 pipeThrough 之后又尝试从原始流取 reader，就会遇到锁定冲突。需要分成两路时，应先明确使用 tee() 等分支策略，而不是绕过管道的所有权约束。

用一个最小转换器确认 transform 的职责

最小的转换器只关心一个问题：输入 chunk 到达时，应该向输出端 enqueue 什么。下面的例子把文本片段转成大写，保留 start、transform 和 flush 三个常见生命周期位置。它没有假设 chunk 是完整句子，所以适合先理解 API 关系。

function createUpperCaseStream() {
  return new TransformStream({
    start(controller) {
      // 这里适合初始化状态；本例不需要预先输出内容。
    },
    transform(chunk, controller) {
      // chunk 只是当前到达的一段数据，不保证对应一条业务消息。
      controller.enqueue(String(chunk).toUpperCase());
    },
    flush(controller) {
      // 输入流正常结束时处理残留状态；本例没有需要刷出的缓存。
    },
  });
}

const resultStream = source.pipeThrough(createUpperCaseStream());

// reader 消费的是转换器的 readable 端，而不是 source 本身。
const reader = resultStream.getReader();
try {
  for (;;) {
    const { value, done } = await reader.read();
    if (done) break;
    renderText(value);
  }
} finally {
  // 无论消费正常结束还是中途抛错，都释放 reader 的锁。
  reader.releaseLock();
}

这里的 controller.enqueue() 可以输出一个 chunk，也可以在一次 transform 中输出多个 chunk。反过来，如果一个输入 chunk 还不够组成目标数据，就暂存到转换器自己的状态中，等后续 chunk 到达再输出。

网络文本为什么必须处理跨 chunk 边界

网络响应通常先是字节流。即使服务端按换行写入，浏览器收到的 chunk 也可能把一行从中间拆开。我的排查顺序是先用 TextDecoderStream 把字节转换成文本，再在自定义 TransformStream 中维护一小段未完成的尾部。

function createLineStream(onLine) {
  let pending = "";

  return new TransformStream({
    transform(chunk, controller) {
      // 当前 chunk 可能从半行开始，先和上次残留内容拼接。
      pending += chunk;
      const lines = pending.split("\n");
      pending = lines.pop() ?? "";

      for (const line of lines) {
        const value = line.replace(/\r$/, "");
        if (!value) continue;
        // 只把已经遇到换行的完整行交给业务回调。
        controller.enqueue(onLine(value));
      }
    },
    flush(controller) {
      // 输入正常结束时，最后一行可能没有换行符，不能静默丢掉。
      const value = pending.replace(/\r$/, "");
      if (value) controller.enqueue(onLine(value));
    },
  });
}

const lines = response.body
  .pipeThrough(new TextDecoderStream())
  .pipeThrough(createLineStream((line) => JSON.parse(line)));

for await (const message of lines) {
  // 每个 message 是按换行边界解析出的对象；解析错误会终止这条管道。
  renderMessage(message);
}

这个例子有两个容易遗漏的点。第一，TextDecoderStream 负责字节到字符串的解码，不负责你的业务分隔符；第二，flush 负责处理最后没有换行符的残留内容。若每行不是 JSON，应该把 JSON.parse 换成实际协议的解析函数，并在异常处保留上下文。

让多个 pipeThrough 组成可读的处理链

多个转换器可以连续连接，因为每次 pipeThrough 都返回新的 ReadableStream。下面的链路只做三件事：检查响应、解码字节、按行转换。最终的 for await...of 消费的是最后一个转换器的输出。

async function readMessages(url, signal) {
  const response = await fetch(url, { signal });
  if (!response.ok || !response.body) {
    // 在建立流式管道前先处理 HTTP 层失败，避免把错误页面当数据解析。
    throw new Error(`流式请求失败：${response.status}`);
  }

  const messageStream = response.body
    .pipeThrough(new TextDecoderStream(), { signal })
    .pipeThrough(createLineStream((line) => JSON.parse(line)), { signal });

  try {
    for await (const message of messageStream) {
      renderMessage(message);
    }
  } catch (error) {
    // 转换器抛错、源流异常或取消都会在消费端汇合到这里。
    reportStreamError(error);
    throw error;
  }
}

链式写法的价值不只是短。它把“字节解码”“消息边界”和“业务消费”分成了可替换的节点，每个节点只承担一种转换职责。若把所有逻辑塞进一个 transform，后续很难判断究竟是字符解码、分隔符处理还是业务解析出了问题。

关闭、异常和取消要按边界决定

pipeThrough(transformStream, options) 的第二个参数不是装饰项，它影响源流、转换器和下游之间的错误传播：

• preventClose: true：源流关闭时，不自动关闭转换目标的 writable 端。
• preventAbort: true：源流出错时，不自动 abort 目标 writable。
• preventCancel: true：目标端出错时，不自动 cancel 源流。
• signal：用 AbortController 取消正在进行的管道。

const controller = new AbortController();

async function consumeWithCancel(source) {
  const transformed = source.pipeThrough(createUpperCaseStream(), {
    // 取消只针对当前管道，不让业务误以为源流一定被永久关闭。
    preventClose: true,
    signal: controller.signal,
  });

  try {
    for await (const chunk of transformed) {
      renderText(chunk);
      if (shouldStop()) {
        // 统一从 AbortController 发出取消信号，避免留下半开的消费任务。
        controller.abort(new Error("用户停止读取"));
        break;
      }
    }
  } finally {
    // 调用方负责释放自己的生命周期对象；不要依赖页面自然销毁。
    controller.abort();
  }
}

不要为了“让错误消失”就把三个 prevent* 全部打开。它们改变的是传播关系：例如下游无法消费时，默认取消源流通常能及时停止网络读取；阻止取消后，源流可能继续产生数据，必须由应用自己负责清理。只有当一条管道连接了多个需要独立管理生命周期的组件时，才有理由改变默认传播。

我最后保留的排查清单

相关问题

pipeThrough 返回的是 TransformStream 吗？

不是。它返回传入转换器的 readable 端，也就是一个新的 ReadableStream，所以可以继续调用 pipeThrough 或最终调用 pipeTo。

transform 每次收到的 chunk 一定是一条完整消息吗？

不一定。chunk 是底层流交付的数据片段，可能半条、整条或多条消息。协议边界需要在转换器中用缓冲区、长度字段或明确分隔符自行维护。

什么时候直接使用 pipeTo？

当转换已经完成，目标是一个最终的 WritableStream 时使用 pipeTo。如果还要改变数据形态或继续接其他处理节点，先用 pipeThrough 保留可读输出。

我现在判断一个 pipeThrough 链是否可靠，只看三件事：每个转换节点是否只做一类工作，是否明确区分 chunk 与业务边界，以及关闭和取消后是否能回收源流。把这三件事写进代码，流式处理就不会只剩下一串看似漂亮的链式调用。