TransformStream 怎样转换流式响应

我第一次把接口改成流式返回时，真正卡住的不是 fetch()，而是“收到一小段数据后，应该在哪里改写它”。答案是把 Response.body 当成输入流，先用 TextDecoderStream 把字节转成文本，再用 TransformStream 在管道中逐块处理，最后交给异步消费者。这样不用等完整响应到齐，也能保持流式消费。

官方资料：https://developer.mozilla.org/en-US/docs/Web/API/TransformStream

下面的示例只做一个边界清楚的任务：把文本响应按块转成大写。生产代码还要处理响应状态、取消信号、跨 chunk 的半行文本和转换异常。

先分清 writable、transform 和 readable

TransformStream 不是一个只接收字符串的函数，而是一对彼此连接的流：writable 是输入端，转换器的 transform(chunk, controller) 负责处理输入，readable 是输出端。写入的数据不会凭空消失，而是由转换器决定是否通过 controller.enqueue() 产生新 chunk。

放进 ReadableStream.pipeThrough() 后，调用方继续拿到的是转换后的 readable。这也是最容易弄反的地方：不要把 TransformStream 本身当成最终读取器，真正消费的是管道返回的可读流。

把响应字节解码后接入 TransformStream

网络响应体通常先表现为字节流，所以文本场景要先解码。下面的转换器按 chunk 处理字符串，再把处理结果送出：

// 把每个文本 chunk 转成大写，并保持流式输出
function upperCaseStream() {
  return new TransformStream({
    transform(chunk, controller) {
      // 处理失败时让管道进入错误状态，避免静默丢数据
      if (typeof chunk !== "string") {
        controller.error(new TypeError("输入必须是文本 chunk"));
        return;
      }
      controller.enqueue(chunk.toUpperCase());
    },
  });
}

async function readTransformedResponse(url, signal) {
  const response = await fetch(url, { signal });
  // 先检查 HTTP 状态，再读取 body，错误更容易定位
  if (!response.ok) {
    throw new Error(`HTTP ${response.status}`);
  }
  if (!response.body) {
    throw new Error("响应没有可读取的 body");
  }

  const stream = response.body
    // 浏览器响应体先从 Uint8Array 解码为字符串
    .pipeThrough(new TextDecoderStream())
    // 转换后的 readable 继续交给异步迭代器
    .pipeThrough(upperCaseStream(), { signal });

  for await (const chunk of stream) {
    // 这里可以把 chunk 增量写入页面或消息缓冲区
    console.log(chunk);
  }
}

这条链的顺序很重要：TextDecoderStream 解决编码边界，upperCaseStream() 解决业务转换，for await 负责消费结果。pipeThrough() 会把当前可读流锁定到这条管道上，因此不要再给同一个 response.body 另取一个 reader。

跨 chunk 数据要交给 lineBuffer

chunk 不是消息边界。服务端发送一行 JSON 或一条事件时，换行符可能被拆到下一个 chunk。若在每个 transform() 调用里直接 split("\n")，就会把半行当成完整消息。

更稳妥的做法是保留未完成的尾部，遇到完整换行时才输出。流关闭前再由 flush() 处理最后一段：

// 将可能跨 chunk 的文本按完整换行拆开
function lineStream(onLine) {
  let lineBuffer = "";

  return new TransformStream({
    transform(chunk, controller) {
      // 尾部暂存起来，避免半行被提前交给业务层
      lineBuffer += chunk;
      const lines = lineBuffer.split("\n");
      lineBuffer = lines.pop() ?? "";

      for (const line of lines) {
        if (line.trim() !== "") {
          controller.enqueue(onLine(line));
        }
      }
    },
    flush(controller) {
      // writable 关闭时，最后没有换行的内容仍然需要处理
      if (lineBuffer.trim() !== "") {
        controller.enqueue(onLine(lineBuffer));
      }
    },
  });
}

如果转换逻辑需要异步操作，可以让 transform() 返回 Promise；但不要在内部无限积压任务。管道会依据消费者速度施加背压，消费者变慢时，上游读取也会受到约束。

生产环境要把错误和取消放在管道边界

网络请求的取消应由调用方持有 AbortController，并把它的 signal 传给 fetch 与 pipeThrough。这样用户离开页面或切换会话时，未完成的读取和转换不会继续占用资源。

// 页面销毁或用户切换请求时调用 controller.abort()
const controller = new AbortController();

readTransformedResponse("/api/stream", controller.signal)
  .catch((error) => {
    // AbortError 属于主动取消，不必当成服务端故障上报
    if (error.name === "AbortError") return;
    console.error("流式响应转换失败", error);
  });

// 取消 fetch、pipeThrough 和后续消费
// controller.abort();

需要区分三类错误：HTTP 非 2xx 是响应状态问题，controller.error() 是转换器无法处理输入，AbortError 则通常是调用方主动取消。把三者混成一个“读取失败”，排查时很难判断应该重试、修复数据还是结束任务。

什么时候适合用 TransformStream

浏览器支持方面，MDN 将 TransformStream 和 pipeThrough() 标为广泛可用，但具体项目仍应按目标浏览器矩阵确认；兼容环境不足时，可以保留一次性读取或服务端预处理作为降级路径。最关键的判断不是“能不能 new”，而是转换任务是否尊重 chunk 边界、错误传播和背压。

常见问题

为什么 transform 收到的不是一整条消息？

流的 chunk 由底层生产者和队列决定，不等同于业务消息。需要完整行、完整事件或完整 JSON 时，必须自己维护缓存和结束条件。

pipeThrough 后还能继续读取原始 response.body 吗？

通常不能。管道会锁定源流，应该消费 pipeThrough() 返回的可读流；如果要分发给多个消费者，需要在设计阶段明确 tee 或独立数据源。

总结：把 Response.body 看成字节输入，把 TextDecoderStream 放在编码边界，把 TransformStream 放在业务转换边界，再用 flush()、AbortSignal 和错误传播补齐生命周期。这样转换流式响应时，既能增量处理，也不会把 chunk 误当成消息。