企业官网建设流程全解析

Rivet Actors 原生 WebSocket 集成实战：基于 Native WebSockets 示例实现实时双向通信

【免费下载链接】actorsRivet Actors are the primitive for stateful workloads. Built for AI agents, collaborative apps, and durable execution.项目地址: https://gitcode.com/GitHub_Trending/riv/actors

导读

本文围绕仓库中的 examples/native-websockets 示例，完整讲解如何在 Rivet Actors 中使用原生 WebSocket API 实现客户端与 Actor 之间的实时双向通信。示例以「多人实时光标房间」为场景，覆盖 Actor 端onWebSocket处理器编写、连接管理与广播、React 前端接入，以及 SDK 底层如何将 WebSocket 桥接到 Actor 运行时。读完本文，你将掌握在 Rivet Actors 上实现自定义 WebSocket 协议、管理连接生命周期并做事件广播的完整实战方案。

示例概览：一个多人实时光标房间

native-websockets示例演示了 Rivet Actors 最核心的实时通信能力：

• 原生 WebSocket 支持：使用标准 WebSocket API（而非 RivetKit 封装的高层连接 API）进行实时通信；
• 双向消息：客户端与 Actor 之间通过websocket.send()与message事件自由收发消息；
• 连接管理：在onWebSocket处理器中通过open/close事件跟踪连接状态，配合onBeforeConnect、onConnect、onDisconnect生命周期钩子；
• 事件广播：一个客户端发来的光标位置更新会被 Actor 推送给所有已连接的 WebSocket 客户端，实现多人实时协作。

整个示例只有 4 个核心文件，非常适合作为学习低层 WebSocket 接入的起点：

快速开始

按照 README 中的步骤即可在本机把示例跑起来（仓库内路径为examples/native-websockets）：

git clone https://github.com/rivet-dev/rivet.git cd rivet/examples/native-websockets npm install npm run dev

打开 package.json 可以看到dev脚本的定义：

"dev": "concurrently -n server,vite \"tsx --watch src/index.ts\" \"vite\""

它使用concurrently同时启动两个进程：

1. server：tsx --watch src/index.ts直接运行 Actor 注册表（registry.start()），并在本地暴露 RivetKit 开发网关（默认http://localhost:6420，见 frontend/App.tsx 中的rivetUrl）；
2. vite：启动 React 前端开发服务器。

前端开发服务器通过 vite.config.ts 将/actors、/metadata、/health代理到http://localhost:6420，其中/actors开启了ws: true以支持 WebSocket 升级：

server: { clearScreen: false, proxy: { "/actors": { target: "http://localhost:6420", ws: true }, "/metadata": { target: "http://localhost:6420" }, "/health": { target: "http://localhost:6420" }, }, },

其余常用脚本：dev:server只跑 Actor 服务端，start以非 watch 模式启动，test运行 vitest，build构建前端。示例依赖rivetkit与@rivetkit/react（版本均为^2.3.17，见 package.json），并且显式安装了ws与@types/ws用于 WebSocket 类型支持。

核心实现：Actor 端的 onWebSocket 处理器

示例的 Actor 定义在 src/index.ts，README 中提到的src/backend/registry.ts在本仓库对应实际文件即src/index.ts。核心是一个名为cursorRoom的 Actor：

import { actor, type RivetMessageEvent, setup, type UniversalWebSocket } from "rivetkit"; interface Vars { websockets: Map<string, UniversalWebSocket>; } export const cursorRoom = actor({ state: { cursors: {} as Record<string, { userId: string; x: number; y: number; timestamp: number }>, }, createVars: (): Vars => { return { websockets: new Map() }; }, actions: { getOrCreate: () => { return { status: "ok" }; }, }, onWebSocket: async (c, websocket: UniversalWebSocket) => { // ...连接处理逻辑 }, });

这段代码展示了 Rivet Actors 定义 Actor 的四个关键组成部分：

• state：持久化的 Actor 状态。这里用cursors记录每个userId的最新光标坐标，状态变更会被 Rivet 运行时持久化，因此即使 Actor 休眠后再唤醒，已连接用户的光标数据仍然保留；
• createVars：创建非持久化的运行时变量。websocketsMap 存放当前存活的所有 WebSocket 连接（userId -> websocket），它是内存态，不参与持久化；
• actions：可被前端通过 RPC 调用的动作。getOrCreate用于前端先解析出房间 Actor 的 ID，再基于该 ID 建立 WebSocket 连接；
• onWebSocket：低层 WebSocket 处理器，接收 Actor 上下文c与一个UniversalWebSocket对象，这是本次示例的核心。

连接建立：校验请求并登记连接

onWebSocket的第一段逻辑负责校验请求参数并把连接登记到vars：

onWebSocket: async (c, websocket: UniversalWebSocket) => { // 从 query 参数中提取 userId if (!c.request) { websocket.close(1008, "Missing request"); return; } const url = new URL(c.request.url); const userId = url.searchParams.get("userId"); // 校验 userId 是否存在 if (!userId) { websocket.close(1008, "Missing userId query parameter"); return; } console.log(`websocket connected: userId=${userId}, actorId=${c.actorId}`); // 把 websocket 存入 vars（非持久化） c.vars.websockets.set(userId, websocket); // 向新连接发送初始状态 websocket.send( JSON.stringify({ type: "init", data: { cursors: c.state.cursors }, }), ); // ... }

几个值得注意的细节：

• c.request是底层 HTTP 升级请求：onWebSocket对应的握手请求可通过c.request访问，从而读取路径与 query 参数。这里正是通过new URL(c.request.url)拿到userId实现「同一 Actor 服务多个用户」的连接标识；
• 关闭码 1008：当请求缺失或userId缺失时，用websocket.close(1008, "...")主动拒绝连接，1008 表示「违反策略」；
• vars与state的分工：WebSocket 连接对象不能序列化，因此放在非持久化的vars中；而光标坐标需要跨休眠周期保留，放在持久化的state中。这种「连接态进 vars、业务态进 state」的划分是从源码结构看出的通用最佳实践。

双向消息：message 事件与广播

建立连接后，Actor 通过addEventListener("message", ...)监听客户端消息，并根据消息type分发处理：

websocket.addEventListener("message", (event: RivetMessageEvent) => { try { const message = JSON.parse(event.data as string); switch (message.type) { case "updateCursor": { const { x, y } = message.data; // 更新持久化状态 c.state.cursors[userId] = { userId, x, y, timestamp: Date.now() }; // 广播给所有 websocket for (const ws of c.vars.websockets.values()) { ws.send(JSON.stringify({ type: "cursorUpdate", data: c.state.cursors[userId] })); } break; } case "getCursors": { // 把当前完整光标状态回给请求方 websocket.send( JSON.stringify({ type: "cursorsState", data: { cursors: c.state.cursors } }), ); break; } } } catch (error) { console.error("error handling websocket message:", error); } });

这里演示了两种消息模式：

• updateCursor：写入 + 广播。先更新持久化状态，再遍历c.vars.websockets向所有连接推送cursorUpdate，实现多人实时同步。这是「事件广播」特性的直接体现；
• getCursors：按需拉取。只把完整状态返回给发起请求的单个连接，用于新客户端主动刷新全量数据。

注意RivetMessageEvent类型由rivetkit导出（见 rivetkit-typescript/packages/rivetkit/src/common/websocket-interface.ts），它的data字段承载实际负载。

连接关闭：清理登记

websocket.addEventListener("close", () => { console.log(`websocket disconnected: userId=${userId}`); c.vars.websockets.delete(userId); });

连接关闭时从vars.websockets中移除对应条目，避免向已断开连接继续发送数据。这就是 README 所述「用onConnect与onDisconnect钩子管理连接」在低层 WebSocket 中的对应实现——根据 docs/content/docs/websocket-handler.mdx 的说明，onWebSocket处理器本身会触发onBeforeConnect、onConnect与onDisconnect生命周期钩子，因此上层连接管理钩子与低层open/close事件是协同工作的。

前端接入：从 getOrCreate 到原始 WebSocket URL

前端逻辑在 frontend/App.tsx，使用@rivetkit/react的createClient连接本地开发网关：

const rivetUrl = "http://localhost:6420"; const client = createClient<typeof registry>(rivetUrl);

建立 WebSocket 连接分三步：

第一步：解析 Actor ID。通过getOrCreate动作拿到（或创建）名为main的房间 Actor：

const actorId = await client.cursorRoom.getOrCreate("main").resolve();

第二步：拼接原始 WebSocket URL：

const wsOrigin = rivetUrl.replace(/^http/, "ws"); const wsUrl = `${wsOrigin}/gateway/${actorId}/raw/websocket?userId=${encodeURIComponent(userId)}`;

URL 结构为ws://localhost:6420/gateway/{actorId}/raw/websocket，其中：

• /gateway/{actorId}是 RivetKit 开发网关的 Actor 路由前缀；
• /raw/websocket指向 Actor 的低层 WebSocket 处理器；
• ?userId=...作为 query 参数传给onWebSocket内的c.request，用于标识连接身份。

第三步：使用标准 WebSocket API 连接：

websocket = new WebSocket(wsUrl); websocket.onopen = () => { console.log("websocket connected"); setConnected(true); }; websocket.onmessage = (event) => { const message = JSON.parse(event.data); switch (message.type) { case "init": // 服务端推送的初始状态 setCursors(message.data.cursors); break; case "cursorUpdate": // 其他用户的光标移动 setCursors((prev) => ({ ...prev, [message.data.userId]: message.data })); break; case "cursorsState": // 全量光标状态 setCursors(message.data.cursors); break; } }; websocket.onclose = () => setConnected(false); websocket.onerror = (error) => console.error("websocket error:", error);

组件卸载时会在useEffect的 cleanup 中关闭连接。鼠标移动时，前端把光标坐标序列化后通过ws.send发往 Actor：

const handleMouseMove = (e: React.MouseEvent<HTMLDivElement>) => { if (ws && ws.readyState === WebSocket.OPEN) { const rect = e.currentTarget.getBoundingClientRect(); ws.send( JSON.stringify({ type: "updateCursor", data: { x: e.clientX - rect.left, y: e.clientY - rect.top }, }), ); } };

前端还按userId哈希分配 8 种颜色之一（CURSOR_COLORS），并用 SVG 渲染光标图形与用户名标签，供多人同时打开页面观察实时同步效果。

自定义 WebSocket 消息协议

示例在前端与 Actor 之间定义了一套基于 JSON 的轻量消息协议，按type字段区分消息语义：

这套协议模式可以平滑扩展：新增消息类型只需在 Actor 的switch (message.type)与前端switch (message.type)中同步增加分支。仓库中更完整的协议示例可参考 rivetkit-typescript/packages/rivetkit/fixtures/driver-test-suite/raw-websocket.ts，它演示了ping/pong、getStats、indexedEcho、二进制消息回显、ctx.sleep()触发休眠等多种消息处理模式。

源码级原理：UniversalWebSocket 与连接桥接

示例中的onWebSocket(c, websocket)拿到的websocket参数类型是UniversalWebSocket。从 SDK 源码看，它由 rivetkit-typescript/packages/rivetkit/src/common/websocket-interface.ts 从@rivetkit/virtual-websocket包重新导出，包括：

export type { RivetCloseEvent, RivetEvent, RivetMessageEvent, UniversalWebSocket } from "@rivetkit/virtual-websocket";

这意味着 Actor 侧使用的是虚拟 WebSocket实现，而非 Node 原生ws。虚拟化带来两个关键能力：

1. WinterTC 兼容：UniversalWebSocket遵循 WinterTC 标准 WebSocket 接口（addEventListener/send/close/readyState等），因此可以无缝对接依赖标准 WebSocket 对象的第三方库（参见 docs/content/docs/websocket-handler.mdx 中的 "WinterTC Compliance" 一节）；
2. 可恢复/可休眠：虚拟 WebSocket 使连接状态能够与 Actor 的休眠周期解耦——即使 Actor 休眠，连接依然由网关保持，消息到达时再唤醒 Actor。

连接桥接的核心实现在 rivetkit-typescript/packages/rivetkit/src/common/inline-websocket-adapter.ts 的InlineWebSocketAdapter类中。它创建一对互相关联的虚拟 WebSocket 对象（clientWs与actorWs）：

• 客户端侧的send()会触发 Actor 侧的message事件（并经dispatchClientMessageWithMetadata附带rivetMessageIndex传输元数据）；
• Actor 侧的send()会触发客户端侧的message事件；
• 任一侧调用close()都会统一走#close(code, reason)，先触发握手层onClose，再把close事件广播给两侧。

初始化时适配器会先把readyState置为 OPEN，再调用处理器（onRestore或onOpen），最后补发连接建立前缓存的客户端消息（pendingClientMessages），从而保证「连接尚未完全就绪时客户端已发出的消息不丢失」。这解释了为什么示例中onWebSocket里立即websocket.send(init)是安全的。

连接方式对比与补充

除示例使用的「手动拼接/gateway/{actorId}/raw/websocket」之外，docs/content/docs/websocket-handler.mdx 还提供了另外两种官方接入方式，可用于不同场景：

• RivetKit Client 的.webSocket()方法：client.chat.getOrCreate(["my-chat"])拿到 actor handle 后调用actor.webSocket("/")，返回标准 WebSocket，代码更简洁；
• .getGatewayUrl()获取网关地址：await actor.getGatewayUrl()后把http(s)替换为ws(s)并拼接/websocket/路径，适合配合wscat等外部工具调试；
• 生产网关地址格式：wss://api.rivet.dev/gateway/{actorId}@{token}/websocket/{...path}，其中{...path}会透传给onWebSocket，可用于在同一个 Actor 内按路径路由到不同功能（如/admin）。

此外，若需要自定义鉴权或连接前置处理，可以在自有服务器上用 RivetKit Client 代理 WebSocket 转发到 Actor（onWebSocket同样会触发onBeforeConnect、onConnect、onDisconnect钩子，详见 docs/content/docs/lifecycle.mdx 与 docs/content/docs/connections.mdx）。低层 WebSocket 的conn.send/c.broadcast不生效，应直接使用websocket.send()（docs/content/docs/connections.mdx）。

测试验证

示例自带一个基于 vitest 的最小集成测试 tests/websocket.test.ts：

import { setupTest } from "rivetkit/test"; import { expect, test } from "vitest"; import { registry } from "../src/index.ts"; test("Cursor room can be created and initialized", async (ctx: any) => { const { client } = await setupTest(ctx, registry); const room = client.cursorRoom.getOrCreate(["test-room"]); // 验证 getOrCreate action 正常工作 const result = await room.getOrCreate(); expect(result).toEqual({ status: "ok" }); });

setupTest来自rivetkit/test，它会基于传入的registry启动一个测试用运行时，并返回类型安全的client，无需真实部署即可验证 Actor 的 RPC 能力。运行npm test即可执行。

对于更完整的低层 WebSocket 行为验证，SDK 自带的驱动测试套件 rivetkit-typescript/packages/rivetkit/tests/driver/raw-websocket.test.ts 提供了waitForJsonMessage、waitForMatchingJsonMessages等辅助函数，覆盖欢迎消息、消息回显、休眠恢复、可休眠 WebSocket 缓冲阈值（HIBERNATABLE_WEBSOCKET_BUFFERED_MESSAGE_SIZE_THRESHOLD）等场景，可作为编写更复杂 WebSocket 测试的参考。

部署提示

示例根目录的 vercel.json 仅声明{ "framework": "hono" }，表明该示例的前端部分可按 Hono 应用部署。生产环境部署时，WebSocket 地址需要替换为实际部署的网关地址（wss://...），并对 WebSocket 连接做鉴权（如网关地址中的{actorId}@{token}格式），相关说明见 docs/content/docs/websocket-handler.mdx 的 HTTP API 一节。

总结

native-websockets示例完整演示了 Rivet Actors 低层 WebSocket 的接入闭环：Actor 端用onWebSocket接收标准 WebSocket 对象、读写持久化状态、按消息类型分发并广播；前端通过getOrCreate解析 Actor ID、拼接/gateway/{actorId}/raw/websocketURL 后使用原生WebSocketAPI 通信；SDK 底层则通过UniversalWebSocket与InlineWebSocketAdapter保证 WinterTC 兼容、消息顺序与休眠恢复能力。掌握这套模式后，你可以在 Rivet Actors 之上自由实现聊天、协作光标、实时看板等任意自定义 WebSocket 协议。

【免费下载链接】actorsRivet Actors are the primitive for stateful workloads. Built for AI agents, collaborative apps, and durable execution.项目地址: https://gitcode.com/GitHub_Trending/riv/actors