WebSocket

Bun.serve() 支持服务器端 WebSocket，具有即时压缩、TLS 支持和 Bun 原生的发布-订阅 API。

⚡️ 7 倍吞吐量Bun 的 WebSocket 速度很快。对于一个在 Linux x64 上的简单聊天室，Bun 每秒可处理的请求数量是 Node.js + "ws" 的 7 倍。

每秒发送消息数	运行时	客户端数
~700,000	(`Bun.serve`) Bun v0.2.1 (x64)	16
~100,000	(`ws`) Node v18.10.0 (x64)	16

Bun 的 WebSocket 实现内部构建于 uWebSockets 之上。

启动 WebSocket 服务器

以下使用 Bun.serve 构建的服务器，会在 fetch 处理器中将每个传入的请求升级为 WebSocket 连接。套接字处理器在 websocket 参数中声明。

server.ts

Bun.serve({
  fetch(req, server) {
    // 将请求升级为 WebSocket
    if (server.upgrade(req)) {
      return; // 不返回 Response
    }
    return new Response("升级失败", { status: 500 });
  },
  websocket: {}, // 处理器
});

Bun 支持以下 WebSocket 事件处理器：

server.ts

Bun.serve({
  fetch(req, server) {}, // 升级逻辑
  websocket: {
    message(ws, message) {}, // 收到消息
    open(ws) {}, // 套接字已打开
    close(ws, code, message) {}, // 套接字已关闭
    drain(ws) {}, // 套接字已准备好接收更多数据
  },
});

为速度设计的 API

在 Bun 中，处理器每个服务器只声明一次，而不是每个套接字声明一次。你向 Bun.serve() 传递一个包含 open、message、close、drain 和 error 方法的单个 WebSocketHandler 对象。这与客户端 WebSocket 类不同，后者扩展了 EventTarget（onmessage、onopen、onclose）。客户端通常打开的套接字连接较少，因此基于事件的 API 在那时是有意义的。但服务器通常打开很多套接字连接，这意味着：

为每个连接添加/移除事件监听器的时间累积起来
存储每个连接的回调函数引用占用额外内存
通常，人们为每个连接创建新函数，这也意味着更多内存

在每个连接间重用同一个处理器对象可以避免这两种开销。

每个处理器的第一个参数是处理事件的 ServerWebSocket 实例。ServerWebSocket 类是一个快速的、Bun 原生的 WebSocket 实现，具有一些额外功能。

server.ts

Bun.serve({
  fetch(req, server) {}, // 升级逻辑
  websocket: {
    message(ws, message) {
      ws.send(message); // 把消息回显回去
    },
  },
});

发送消息

每个 ServerWebSocket 实例都有一个 .send() 方法，用于向客户端发送消息。它支持多种输入类型。

server.ts

Bun.serve({
  fetch(req, server) {}, // 升级逻辑
  websocket: {
    message(ws, message) {
      ws.send("Hello world"); // string
      ws.send(response.arrayBuffer()); // ArrayBuffer
      ws.send(new Uint8Array([1, 2, 3])); // TypedArray | DataView
    },
  },
});

头部

升级成功后，Bun 按照规范发送 101 Switching Protocols 响应。要为该 Response 附加额外的 headers，将它们传递给 server.upgrade()。

server.ts

Bun.serve({
  fetch(req, server) {
    const sessionId = await generateSessionId();
    server.upgrade(req, {
      headers: { 
        "Set-Cookie": `SessionId=${sessionId}`, 
      }, 
    });
  },
  websocket: {}, // 处理器
});

上下文数据

在 .upgrade() 调用中向新的 WebSocket 附加上下文 data。它可以在 WebSocket 处理器内部的 ws.data 属性上访问。要强类型化 ws.data，向 websocket 处理器对象添加一个 data 属性。这会为所有生命周期钩子中的 ws.data 提供类型。

server.ts

type WebSocketData = {
  createdAt: number;
  channelId: string;
  authToken: string;
};

Bun.serve({
  fetch(req, server) {
    const cookies = new Bun.CookieMap(req.headers.get("cookie")!);

    server.upgrade(req, {
      // 此对象必须符合 WebSocketData
      data: {
        createdAt: Date.now(),
        channelId: new URL(req.url).searchParams.get("channelId"),
        authToken: cookies.get("X-Token"),
      },
    });

    return undefined;
  },
  websocket: {
    // TypeScript：像这样指定 ws.data 的类型
    data: {} as WebSocketData,
    // 收到消息时调用的处理器
    async message(ws, message) {
      // ws.data 现在被正确类型化为 WebSocketData
      const user = getUserFromToken(ws.data.authToken);

      await saveMessageToDatabase({
        channel: ws.data.channelId,
        message: String(message),
        userId: user.id,
      });
    },
  },
});

之前，你可以通过在 Bun.serve 上使用类型参数来指定 ws.data 的类型，例如 Bun.serve<MyData>({...})。由于 TypeScript 的一个限制，这种模式已被移除，改为使用 data 属性。

要从浏览器连接到此服务器，创建一个新的 WebSocket。

browser.js

const socket = new WebSocket("ws://localhost:3000/chat");

socket.addEventListener("message", event => {
  console.log(event.data);
});

识别用户页面上设置的 Cookie 会随 WebSocket 升级请求一起发送，并可在 fetch 处理器的 req.headers 中获取。解析它们以识别连接中的用户，并相应地设置 data。

发布/订阅

Bun 的 ServerWebSocket 包含一个原生的发布-订阅 API，用于基于主题的广播。单个套接字可以 .subscribe() 到一个主题（通过字符串标识符指定），并向该主题的所有其他订阅者 .publish() 消息（不包括自身）。这种基于主题的广播 API 类似于 MQTT 和 Redis Pub/Sub。

server.ts

const server = Bun.serve({
  fetch(req, server) {
    const url = new URL(req.url);
    if (url.pathname === "/chat") {
      console.log(`升级!`);
      const username = getUsernameFromReq(req);
      const success = server.upgrade(req, { data: { username } });
      return success ? undefined : new Response("WebSocket 升级错误", { status: 400 });
    }

    return new Response("Hello world");
  },
  websocket: {
    // TypeScript：像这样指定 ws.data 的类型
    data: {} as { username: string },
    open(ws) {
      const msg = `${ws.data.username} 进入了聊天室`;
      ws.subscribe("the-group-chat");
      server.publish("the-group-chat", msg);
    },
    message(ws, message) {
      // 这是一个群聊
      // 所以服务器将传入的消息重新广播给每个人
      server.publish("the-group-chat", `${ws.data.username}: ${message}`);

      // 查看当前订阅
      console.log(ws.subscriptions); // ["the-group-chat"]
    },
    close(ws) {
      const msg = `${ws.data.username} 离开了聊天室`;
      ws.unsubscribe("the-group-chat");
      server.publish("the-group-chat", msg);
    },
  },
});

console.log(`监听 ${server.hostname}:${server.port}`);

调用 .publish(data) 会将消息发送给某个主题的所有订阅者，除了调用 .publish() 的套接字本身。要向某个主题的所有订阅者发送消息，请使用 Server 实例上的 .publish() 方法。

const server = Bun.serve({
  websocket: {
    // ...
  },
});

// 监听某个外部事件
server.publish("the-group-chat", "Hello world");

压缩

使用 perMessageDeflate 参数启用每条消息的压缩。

server.ts

Bun.serve({
  websocket: {
    perMessageDeflate: true, 
  },
});

要压缩单条消息，将 boolean 作为第二个参数传递给 .send()。

ws.send("Hello world", true);

有关压缩特性的精细控制，请参阅参考。

背压

ServerWebSocket 的 .send(message) 方法返回一个 number，表示操作的结果。

-1 — 消息已入队但存在背压
0 — 由于连接问题，消息被丢弃
1+ — 发送的字节数

超时和限制

默认情况下，Bun 会关闭空闲 120 秒的 WebSocket 连接。使用 idleTimeout 参数进行配置。

Bun.serve({
  fetch(req, server) {}, // 升级逻辑
  websocket: {
    idleTimeout: 60, // 60 秒
  },
});

Bun 也会在收到大于 16 MB 的消息时关闭 WebSocket 连接。使用 maxPayloadLength 参数进行配置。

Bun.serve({
  fetch(req, server) {}, // 升级逻辑
  websocket: {
    maxPayloadLength: 1024 * 1024, // 1 MB
  },
});

连接到 `Websocket` 服务器

Bun 实现了 WebSocket 类。要创建一个连接到 ws:// 或 wss:// 服务器的 WebSocket 客户端，创建一个 WebSocket 实例，就像在浏览器中一样。

const socket = new WebSocket("ws://localhost:3000");

// 使用子协议协商
const socket2 = new WebSocket("ws://localhost:3000", ["soap", "wamp"]);

在浏览器中，页面上设置的 Cookie 会随 WebSocket 升级请求一起发送。这是 WebSocket API 的标准功能。在 Bun 中，你也可以直接在构造函数中设置自定义头部。这是 Bun 对 WebSocket 标准的专用扩展。在浏览器中不起作用。

const socket = new WebSocket("ws://localhost:3000", {
  headers: {
    /* 自定义头部 */
  }, 
});

为套接字添加事件监听器：

// 收到消息
socket.addEventListener("message", event => {});

// 套接字已打开
socket.addEventListener("open", event => {});

// 套接字已关闭
socket.addEventListener("close", event => {});

// 错误处理器
socket.addEventListener("error", event => {});

参考

查看 TypeScript 定义

namespace Bun {
  export function serve(params: {
    fetch: (req: Request, server: Server) => Response | Promise<Response>;
    websocket?: {
      message: (ws: ServerWebSocket, message: string | ArrayBuffer | Uint8Array) => void;
      open?: (ws: ServerWebSocket) => void;
      close?: (ws: ServerWebSocket, code: number, reason: string) => void;
      error?: (ws: ServerWebSocket, error: Error) => void;
      drain?: (ws: ServerWebSocket) => void;

      maxPayloadLength?: number; // 默认: 16 * 1024 * 1024 = 16 MB
      idleTimeout?: number; // 默认: 120 (秒)
      backpressureLimit?: number; // 默认: 16 * 1024 * 1024 = 16 MB
      closeOnBackpressureLimit?: boolean; // 默认: false
      sendPings?: boolean; // 默认: true
      publishToSelf?: boolean; // 默认: false

      perMessageDeflate?:
        | boolean
        | {
            compress?: boolean | Compressor;
            decompress?: boolean | Compressor;
          };
    };
  }): Server;
}

type Compressor =
  | `"disable"`
  | `"shared"`
  | `"dedicated"`
  | `"3KB"`
  | `"4KB"`
  | `"8KB"`
  | `"16KB"`
  | `"32KB"`
  | `"64KB"`
  | `"128KB"`
  | `"256KB"`;

interface Server {
  pendingWebSockets: number;
  publish(topic: string, data: string | ArrayBufferView | ArrayBuffer, compress?: boolean): number;
  upgrade(
    req: Request,
    options?: {
      headers?: HeadersInit;
      data?: any;
    },
  ): boolean;
}

interface ServerWebSocket {
  readonly data: any;
  readonly readyState: number;
  readonly remoteAddress: string;
  readonly subscriptions: string[];
  send(message: string | ArrayBuffer | Uint8Array, compress?: boolean): number;
  close(code?: number, reason?: string): void;
  subscribe(topic: string): void;
  unsubscribe(topic: string): void;
  publish(topic: string, message: string | ArrayBuffer | Uint8Array): void;
  isSubscribed(topic: string): boolean;
  cork(cb: (ws: ServerWebSocket) => void): void;
}

开始入门

核心运行时

文件与模块系统

HTTP 服务器

网络通信

数据与存储

并发

进程与系统

互操作与工具

实用工具

标准与兼容性

贡献指南

WebSocket

启动 WebSocket 服务器

发送消息

头部

上下文数据

发布/订阅

压缩

背压

超时和限制

连接到 `Websocket` 服务器

参考

​启动 WebSocket 服务器

​发送消息

​头部

​上下文数据

​发布/订阅

​压缩

​背压

​超时和限制

​连接到 Websocket 服务器

​参考

启动 WebSocket 服务器

发送消息

头部

上下文数据

发布/订阅

压缩

背压

超时和限制

连接到 `Websocket` 服务器

参考