WebSocketStream怎样用 Streams API 处理背压

如果实时消息的到达速度超过页面处理速度，普通 WebSocket 客户端很容易把未处理消息越积越多。WebSocketStream 的思路是把连接接到 Streams API：接收侧暴露 ReadableStream，发送侧暴露 WritableStream，浏览器根据流的处理能力自动调节读写节奏。

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

本文用一个“实时订阅客户端”串起这套关系。先说明边界：MDN 当前将 WebSocketStream 标为实验性、非标准技术，浏览器支持范围有限；它适合用来理解 Streams API 背压和做受控实验，不应直接当成所有浏览器都能使用的生产默认方案。

先把背压理解成一条反向信号

Streams API 中，数据从源头流向消费者。消费者暂时处理不过来时，后面的流会把“先慢一点”的信号传回前面的队列或数据源，这就是背压。它解决的不是网络断线，而是生产速度和消费速度不匹配。

因此，背压代码的重点不是自己维护一个无限数组，而是让读取和写入都尊重流提供的 Promise。接收侧处理完当前消息再读取下一条，发送侧等待 writer 可以继续写入后再提交数据，队列就不会被业务代码无界放大。

从 WebSocket 连接拿到两条流

new WebSocketStream(url) 创建的是一个基于 Promise 的连接对象。连接成功后，wss.opened 会给出 readable 和 writable，前者负责接收，后者负责发送。它们分别只能绑定一个活动 reader 或 writer，所以同一条流不要在多个函数里重复获取锁。

async function openChannel(url, signal) {
  // 先做能力判断，避免不支持的浏览器直接抛出构造器错误。
  if (!("WebSocketStream" in self)) {
    throw new Error("当前浏览器不支持 WebSocketStream");
  }

  // signal 可用于在握手尚未完成时取消连接尝试。
  const socket = new WebSocketStream(url, { signal });
  const { readable, writable, protocol } = await socket.opened;

  // 一条连接拆成接收 reader 和发送 writer，分别管理各自的节奏。
  return {
    socket,
    reader: readable.getReader(),
    writer: writable.getWriter(),
    protocol,
  };
}

const channel = await openChannel(
  "wss://example.com/events",
  AbortSignal.timeout(5000), // 握手超过 5 秒就取消
);

示例中的地址只是代码占位符，不代表一个可直接连接的服务。真正接入时，服务端协议、认证方式和消息格式仍要由项目自己的接口约定决定。

接收端不要提前把消息全部读出来

背压最容易被破坏的地方，是为了“提高吞吐”而提前调用很多次 read()，再把结果塞进自己的数组。这样一来，流本身的节奏控制被业务队列绕开了。

更稳妥的做法是：一次只取一条消息，完成解析和业务处理后再进入下一轮。处理函数如果本身是异步的，await 就会自然成为消费速度的边界。

async function consumeMessages(reader, onMessage) {
  try {
    while (true) {
      // 下一次 read 放在处理完成之后，避免人为制造预取队列。
      const { value, done } = await reader.read();
      if (done) {
        break;
      }

      // 服务端可能发送字符串，也可能发送二进制块；按协议选择解析方式。
      const text = typeof value === "string"
        ? value
        : new TextDecoder().decode(value);

      // 处理函数没有完成前，不会继续拉取下一条消息。
      await onMessage(JSON.parse(text));
    }
  } finally {
    // reader 不再使用时释放锁，让关闭流程可以接管这条流。
    reader.releaseLock();
  }
}

await consumeMessages(channel.reader, async (event) => {
  // 这里替换成实际的渲染、落库或业务处理逻辑。
  await renderEvent(event);
});

这段代码没有承诺具体的吞吐量。它只把“应用处理完成”放在下一次读取之前，让 Streams API 有机会把消费端的慢速传回连接链路。

发送端要等待 writer 的安全写入时机

发送端也有同样的原则：不要在一个快速循环里无条件调用 write()。可以先等待 writer.ready，再等待本次 write() 完成。这样业务层不会用自己的发送数组把可写流的队列重新扩大。

async function sendJson(writer, payload) {
  // 等待写入端恢复可写状态，避免连续生产数据压满队列。
  await writer.ready;

  // 每次写入都等待 Promise，保证发送循环服从流的处理节奏。
  await writer.write(JSON.stringify(payload));
}

await sendJson(channel.writer, {
  type: "subscribe",
  topic: "orders",
});

writer.ready 和 write() 解决的是写入时机，不会替你设计重试、去重或业务确认协议。如果消息不能丢，仍要在应用层增加消息 ID、确认和失败恢复。

把读写、关闭和异常放进一个生命周期

一个可复用的客户端至少要处理三类状态：握手失败、流在读取或写入时失败、连接正常关闭。不要只保存 reader 和 writer 而忘记 wss.closed；连接结束后应该释放锁，并按项目约定关闭或取消剩余资源。

async function runClient(url) {
  const controller = new AbortController();
  const { socket, reader, writer } = await openChannel(url, controller.signal);

  try {
    await sendJson(writer, { type: "subscribe", topic: "orders" });
    await consumeMessages(reader, async (event) => {
      // 慢处理会自然限制下一次 read 的进入速度。
      await renderEvent(event);
    });
  } catch (error) {
    // 将网络错误交给上层决定是否重连，避免这里偷偷无限重试。
    console.error("WebSocketStream 生命周期失败", error);
  } finally {
    // 先释放 writer 的锁，再关闭连接，避免遗留不可接管的锁。
    writer.releaseLock();
    socket.close({ closeCode: 1000, reason: "客户端结束订阅" });

    // closed 用于取得服务端最终给出的关闭信息。
    const result = await socket.closed;
    console.info("连接已关闭", result.closeCode, result.reason);
  }
}

await runClient("wss://example.com/events");

示例中的 renderEvent 代表项目自己的异步处理函数。实际项目里还应根据页面生命周期，在组件销毁时主动结束订阅，避免后台页面继续持有连接。

兼容性决定它能不能进入生产

这项 API 的优点是背压已经进入流模型，代码不必围绕 bufferedAmount 自己搭一套发送队列；但代价是标准化和浏览器覆盖范围都不稳定。MDN 目前把它列为实验性、非标准技术，采用前必须查看目标浏览器的兼容性表。

因此，实际落地可以把能力检测放在入口，把 WebSocketStream 封装成一个可替换传输层：支持时走 Streams API，不支持时切换到已有 WebSocket 实现，并在后者内部明确限制队列长度和丢弃策略。

最后记住这三个背压检查点

• 接收侧：不要先批量调用 read()，处理完当前消息再读下一条。
• 发送侧：等待 writer.ready，并等待 writer.write() 的 Promise。
• 产品侧：把实验性 API 的能力检测、降级传输和连接关闭纳入生命周期。

WebSocketStream 的关键不在于把 WebSocket 换一个构造器，而在于让实时数据经过可感知队列的 Streams API。只要读写两端都尊重 Promise 和流锁，背压才会成为系统设计的一部分，而不是消息堆积后的补救代码。

相关问题

WebSocketStream 和 WebSocket 最大区别是什么？

WebSocketStream 以 Promise、ReadableStream 和 WritableStream 暴露连接，能够利用 Streams API 的背压；WebSocket 更成熟，但没有同样的内建背压接口。

只使用 reader.read() 就能保证服务端永远不堆积吗？

不能。它能让客户端消费节奏参与流控制，但服务端自身的缓冲、协议实现和业务队列仍需单独设计。

WebSocketStream 现在适合直接替换线上 WebSocket 吗？

不建议无条件替换。先看目标浏览器支持，再准备 WebSocket 或 WebTransport 等替代路径，并把消息确认与重连策略放在应用层。