用 Rivet Actors 与 Yjs 构建实时协作文档：CRDT 同步、光标 Presence 与 KV 持久化实战

用 Rivet Actors 与 Yjs 构建实时协作文档CRDT 同步、光标 Presence 与 KV 持久化实战【免费下载链接】actorsRivet Actors are the primitive for stateful workloads. Built for AI agents, collaborative apps, and durable execution.项目地址: https://gitcode.com/GitHub_Trending/riv/actors本篇文章围绕仓库中的collaborative-document示例完整拆解如何用 Rivet Actors 实现一个多人在线协作文本编辑器以“每个文档一个 Actor”承载 Yjs CRDT 状态以“Coordinator Actor”索引工作区内的文档通过 Actor 事件广播增量更新与光标 Presence并用 Actor KV 完成崩溃恢复级的持久化。读完本文你将掌握 Coordinator/Data Actor 双层架构、Yjs 二进制更新在 Actor 之间的流转方式以及 React 前端如何通过rivetkit/react无缝接入这套实时链路。示例概览它在演示什么examples/collaborative-document/是一个完整的共享文本编辑器示例其核心组合是Rivet Actors负责有状态工作负载的运行时原语承载 CRDT 文档状态、连接管理与持久化Yjs成熟的 CRDTConflict-Free Replicated Data Type库解决多人同时编辑同一文本时的冲突合并问题y-protocols/awareness在 Yjs 之上提供 Presence在线状态、用户名、光标位置传播协议。从项目描述看Rivet Actors 本身就是为“有状态工作负载”设计的原语适用于 AI Agent、协作应用与持久化执行场景而这个示例恰好演示了其中“协作应用”这一类多人共享同一个文档、实时看到彼此的输入与光标并且刷新/重启后内容不丢失。示例的四大能力在 README.md 中被概括为能力说明对应实现Coordinator 模式每个工作区一个documentListActor索引并管理该工作区下的文档 Actorsrc/index.tsCRDT 同步Document Actor 将 Yjs 增量更新广播给所有协作者src/index.tsPresence 与光标Awareness 更新通过 Actor 事件流动实现实时光标src/index.ts持久化Yjs 快照写入 Actor KV支持崩溃恢复src/index.ts快速开始克隆、安装与运行按照 README 的指引在本地启动整个示例只需三条命令git clone https://github.com/rivet-dev/rivet.git cd rivet/examples/collaborative-document npm install npm run dev在当前仓库中示例位于examples/collaborative-document/其package.json中的核心脚本如下见 package.json{ scripts: { dev: concurrently -n server,vite \tsx --watch src/index.ts\ \vite\, dev:server: tsx --watch src/index.ts, check-types: tsc --noEmit, test: vitest run, build: vite build, start: tsx src/index.ts } }npm run dev通过concurrently同时启动两件事用tsx --watch热重载运行 Actor 服务端入口 src/index.ts以及 Vite 开发服务器提供 React 前端npm run start只运行服务端适合对接外部前端依赖方面服务端使用rivetkit前端使用rivetkit/reactCRDT 部分依赖yjs与y-protocols并引入了react/react-dom与 Vite 工具链。启动后前端页面默认通过http://localhost:6420连接 Actor 服务默认工作区名为design-team默认用户名为Ada见 frontend/App.tsx。架构总览Coordinator Data Actor 双层模型整个示例由两个 Actor 组成形成一个典型的双层结构┌─────────────────────────────┐ │ documentList (Coordinator) │ 每个工作区一个 │ state: DocumentSummary[] │ 负责创建/列出/删除文档 └──────────────┬──────────────┘ │ 调用 document.create(...) ┌────────────────────────┼────────────────────────┐ ▼ ▼ ▼ ┌──────────────────┐ ┌──────────────────┐ ┌──────────────────┐ │ document actor │ │ document actor │ │ document actor │ │ key:[ws, docId] │ │ key:[ws, docId] │ │ key:[ws, docId] │ │ Y.Doc Awareness│ │ Y.Doc Awareness│ │ Y.Doc Awareness│ │ KV: yjs:doc │ │ KV: yjs:doc │ │ KV: yjs:doc │ └──────────────────┘ └──────────────────┘ └──────────────────┘documentListCoordinator Actor一个工作区对应一个实例以工作区 ID 作为 Actor Key它维护该工作区下的文档摘要列表DocumentSummary[]负责创建、列出与删除文档documentData Actor一份文档对应一个实例以[workspaceId, documentId]作为 Actor Key它持有该文档的 YjsY.Doc与Awareness实例是 CRDT 状态与实时广播的载体。这与官方 Design Patterns 文档 中“Coordinator Data Actors”一节完全对应Data actors承载应用主体逻辑聊天室、用户会话、游戏大厅等Coordinator actors则像索引一样追踪其他 Actor聊天室列表、活跃用户列表、游戏大厅列表。文档还强调这种“每个实体一个 Actor”的模式天然具备可扩展性——每个 Actor 的私有状态相互隔离无需锁与共享内存协调负载会自然分散到海量 Actor 上。Document ActorYjs CRDT 状态、广播与持久化documentActor 是示例的核心定义在 src/index.ts。它同时用到了 Rivetkit 提供的连接状态、事件、持久化状态与 KV 存储四类机制。1. 连接级状态connState记录每个连接的 Awareness 客户端export const document actor({ connState: { clientIds: [] as number[], }, // ... });connState是按连接隔离的可变状态每个 WebSocket 连接都持有自己的一份clientIds数组用于记录该连接在 Yjs Awareness 协议中的客户端 ID。这样断开连接时可以精确地只清理属于该连接的光标与在线状态。2. 事件定义sync与awarenessevents: { sync: eventSyncEvent(), awareness: eventAwarenessEvent(), },其中SyncEvent { update: number[] }AwarenessEvent { update: number[] }。两个事件都只携带一个number[]即 Yjs/y-protocols 的二进制更新被转成的字节数组——这是广播的最小载荷。eventT()是 Rivetkit 的强类型事件声明配合 Events 文档 中介绍的c.broadcast(eventName, ...args)与客户端connection.on(...)/useEvent(...)构成完整的实时通道。3.createState可持久化的文档元数据createState: (_c, input: DocumentInput): DocumentState ({ title: input.title, createdAt: input.createdAt, updatedAt: input.createdAt, }),createState在 Actor 首次创建时执行返回的DocumentState会被自动持久化并在重启后恢复见官方 State 文档。这里保存的是轻量元数据标题与时间戳而重量级的 Yjs 文档数据则走 KV。4.createVars从 KV 恢复 Yjs 文档createVars: async (c) { const doc new Y.Doc(); const stored await c.kv.get(yjs:doc, { type: binary }); if (stored) { Y.applyUpdate(doc, stored); } const awareness new Awareness(doc); return { doc, awareness }; },createVars在 Actor 每次重新启动时执行用于构建运行时变量。这里的逻辑是新建Y.Doc从 KV 以binary类型读取键yjs:doc若存在则用Y.applyUpdate把历史快照灌入文档基于该文档创建Awareness实例返回给后续所有 action/事件处理器使用。这也印证了官方 Design Patterns 文档 中“用createVars加载外部状态”的推荐做法把“需要缓存/加载的运行时数据”与“需要持久化的 actor state”分开管理。5. Actions文档读写与增量广播actions: { getContent: (c) { const update Y.encodeStateAsUpdate(c.vars.doc); return toNumbers(update); }, applyUpdate: async ( c, update: number[], kind: UpdateKind, clientId?: number, ) { if (kind sync) { const bytes toBytes(update); Y.applyUpdate(c.vars.doc, bytes, client); const snapshot Y.encodeStateAsUpdate(c.vars.doc); await c.kv.put(yjs:doc, snapshot); c.state.updatedAt Date.now(); c.broadcast(sync, { update }); return; } // ...awareness 分支 }, getAwareness: (c) { const clients Array.from(c.vars.awareness.getStates().keys()); const update encodeAwarenessUpdate(c.vars.awareness, clients); return toNumbers(update); }, },三个 action 各司其职getContent将当前完整的 Yjs 文档状态编码为 update供新加入的协作者做初始同步类似“全量快照”。applyUpdatesync 分支这是写入链路的核心。客户端把本地编辑产生的 Yjs 增量update传进来后Actor 依次执行Y.applyUpdate(c.vars.doc, bytes, client)把增量合并进服务器端的Y.DocY.encodeStateAsUpdate生成最新全量快照写入 KV 键yjs:doc——每次编辑都落盘实现崩溃恢复刷新updatedAt时间戳c.broadcast(sync, { update })把原始增量广播给所有已连接协作者。getAwareness把当前 Awareness 状态编码后返回供新连接恢复在线列表与光标。这种“接收增量 → 合并 → 快照落盘 → 原样广播”的流程正是 CRDT 场景下的典型 Actor 语义Actor 作为所有写入的汇聚点保证了持久化与广播的顺序一致。6.onDisconnect连接断开时清理 PresenceonDisconnect: (c, conn) { const clientIds conn.state.clientIds; if (clientIds.length 0) return; removeAwarenessStates(c.vars.awareness, clientIds, disconnect); const update encodeAwarenessUpdate(c.vars.awareness, clientIds); c.broadcast(awareness, { update: toNumbers(update) }); conn.state.clientIds []; },当某条连接断开时Actor 用之前记录在该连接connState.clientIds中的客户端 ID从 Awareness 中移除对应状态并把“移除”事件广播给其余协作者让他们的在线列表与远程光标立刻消失。DocumentList Coordinator文档的创建、索引与跨 Actor 调用documentListActor 定义在 src/index.ts它只维护一个持久化数组状态state: { documents: [] as DocumentSummary[], },三个 action 构成完整的文档生命周期管理createDocument生成randomUUID()作为文档 ID以c.key[0] ?? default作为工作区 ID然后通过c.clienttypeof registry()跨 Actor 调用document.create([workspaceId, documentId], { input: {...} })并await handle.resolve()等待文档 Actor 创建完成随后把DocumentSummary追加进自己的状态数组并返回listDocuments直接返回c.state.documents供前端初始化文档列表deleteDocument按 ID 从数组中过滤掉对应摘要。createDocument: async (c, title: string) { const documentId randomUUID(); const createdAt Date.now(); const workspaceId c.key[0] ?? default; const safeTitle title.trim() || Untitled document; const client c.clienttypeof registry(); const handle await client.document.create( [workspaceId, documentId], { input: { title: safeTitle, createdAt } }, ); await handle.resolve(); // ... },这里体现了两个重要的 Rivetkit 机制跨 Actor 调用Coordinator 使用c.clienttypeof registry()获得类型安全的客户端句柄再以client.document.create(key, params)的方式按 Key 精确寻址并唤醒目标文档 ActorKey 驱动路由document的 Actor Key 是[workspaceId, documentId]二元组意味着同一个工作区的不同文档、不同工作区的同名文档都是彼此独立的 Actor 实例——文档数量可以随工作区线性扩展这正是 Design Patterns 中“Actor Per Entity Coordinator”模式的扩展性来源。最后两个 Actor 通过setup注册并启动服务src/index.tsexport const registry setup({ use: { document, documentList }, }); registry.start();setup返回的registry类型同时被前端导入用于让rivetkit/react获得完整的类型推导见下文。前端React 编辑器、实时光标与 Presence 面板前端由 frontend/App.tsx 与 frontend/main.tsx 组成入口 HTML 在 index.html。1. 建立类型安全的 Actor 客户端const { useActor } createRivetKittypeof registry(http://localhost:6420);createRivetKittypeof registry把服务端注册表类型桥接到前端useActor({ name: document, key: [workspaceId, document.id] })即可按与后端完全一致的 Key 规则连接到指定文档。连接成功后documentActor.connection会提供getContent()、applyUpdate()、getAwareness()等 RPC 方法以及useEvent(sync, cb)、useEvent(awareness, cb)事件订阅。2. 本地 Y.Doc 与远程更新的双向同步编辑器组件在挂载时创建本地Y.Doc与Awareness并建立两条监听链frontend/App.tsxdoc.on(update, ...)本地编辑产生 Yjs 增量时如果来源不是remote就把增量通过connection.applyUpdate(toNumbers(update), sync)推给 Document Actorawareness.on(update, ...)本地光标/用户名变化时把 Awareness 增量通过connection.applyUpdate(..., awareness, awareness.clientID)推给 Actor。反过来documentActor.useEvent(sync, ...)收到广播时用remote为 origin 调用Y.applyUpdate(doc, ...)合入远程增量从而不会再次回推useEvent(awareness, ...)则合入远程 Presence 并刷新 UIfrontend/App.tsx。3. 初始同步全量拉取内容与在线状态新连接不能只靠事件广播还必须拿到历史内容。组件在连接就绪后并行调用两个 actionfrontend/App.tsxconnection.getContent().then((update) { Y.applyUpdate(doc, toBytes(update), remote); setContent(text.toString()); }); connection.getAwareness().then((update) { applyAwarenessUpdate(awareness, toBytes(update), remote); setPresence(buildPresence(awareness)); });这正好对应后端getContent全量 Yjs 快照与getAwareness全量在线状态两个 action——“进入时全量拉取之后增量广播”是此类实时系统的标准握手协议。4. 光标叠加渲染与 Presence 面板前端对远程光标的渲染做了精细处理每次光标/选区变化时通过awareness.setLocalStateField(cursor, { index, length })更新本地光标状态frontend/App.tsxbuildPresence从awareness.getStates()中提取每个客户端 ID 的user与cursor信息renderWithCursors将文本按光标位置切片在文本层之上叠加一个绝对定位、pointer-events: none的透明覆盖层cursor-overlay用 CSS 变量--cursor-color为每个协作者渲染彩色光标竖线并把用户名以data-name的形式显示在光标上方frontend/App.tsx侧栏presence-panel列出所有在线协作者、各自颜色与光标位置Cursor: N或idle。此外用户颜色由用户名哈希从 8 色调色板CURSOR_COLORS中确定同一用户在不同页面打开时颜色保持一致frontend/App.tsx。开发配置细节Vite 代理与前后端联通前端通过 Vite 开发服务器代理 Actor 的 HTTP 与 WebSocket 流量见 vite.config.tsserver: { clearScreen: false, proxy: { /actors: { target: http://localhost:6420, ws: true }, /metadata: { target: http://localhost:6420 }, /health: { target: http://localhost:6420 }, }, },/actors是 Actor 的 RPC 与实时通道端点ws: true表示同时代理 WebSocketuseActor的持久连接正依赖它/metadata与/health用于服务发现与健康检查clearScreen: false保证concurrently同时输出 server 与 vite 日志时不会互相刷屏。tsconfig.json开启了strict、allowImportingTsExtensions与rewriteRelativeImportExtensions因此 frontend/App.tsx 可以直接import type { registry } from ../src/index.ts把服务端类型安全地复用进前端。关联官方文档把示例映射到 Rivet 能力体系示例中出现的每个机制都可以在仓库的官方文档中找到完整说明示例中的用法官方文档要点documentList索引文档Design PatternsCoordinator Data ActorsCoordinator 负责追踪 Data Actor保持“每实体一 Actor”的粒度c.broadcast(sync, ...)EventsRealtime事件面向.connect()的持久连接广播客户端用connection.on/useEvent订阅actions: { ... }与c.client跨 Actor 调用ActionsAction 是 Actor 的 RPC 入口c.client提供类型安全的跨 Actor 调用c.kv.get/put(yjs:doc)KV二进制键值存储支持type: binary读取createState持久化元数据State状态跨重启保留createVars用于加载运行时缓存如外部数据库数据可复用的设计要点从该示例中提炼出的通用模式适用于任何基于 Rivet Actors 的实时协作系统双层 Actor 架构Coordinator 只做索引与编排轻量、数量少Data Actor 持有完整业务状态重量级、按实体水平扩展。列表页打 Coordinator编辑页打 Data Actor职责清晰互不干扰。二进制载荷直通Yjs 更新本身就是字节流示例直接以number[]形式在 action 参数、事件载荷与 KV 之间传递避免重复编码解码toNumbers/toBytes两个小工具函数贯穿前后端。“全量拉取 增量广播”的握手协议新协作者先getContentgetAwareness对齐基线之后只消费增量事件兼顾了正确性与带宽效率。origin 标签防止回声客户端用remote标记远端来源的更新合入时不再回推避免同步风暴。每次写入即快照落盘applyUpdate在合并增量后立刻把全量快照写入 KV 的yjs:doc键用“慢一点、但绝不会丢”换取崩溃恢复能力对更大规模场景可在此基础上叠加定期压缩或增量存储策略。连接级状态管理 Presence 生命周期connState.clientIds让onDisconnect能精确清理断开连接留下的 Awareness 状态保证在线列表始终反映真实连接。结语collaborative-document是一个麻雀虽小、五脏俱全的参考实现它把 Rivet Actors 的持久化状态、KV、事件广播、跨 Actor 调用与连接生命周期和 Yjs 的 CRDT 收敛、Awareness Presence 协议完整地组装成一个可运行的真实应用。无论你是要构建协作文档、共享白板、多人表格还是 AI Agent 协作界面都可以以这份示例为骨架按需替换编辑器层如 CodeMirror/ProseMirror 与 Yjs 的官方绑定、接入鉴权参考 Connections并把 KV 键扩展为多版本快照以支持历史回溯。【免费下载链接】actorsRivet Actors are the primitive for stateful workloads. Built for AI agents, collaborative apps, and durable execution.项目地址: https://gitcode.com/GitHub_Trending/riv/actors创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考