21 6月 2026 阅读时间 10 分钟 Multica 拆解

Agent 的实时层:Multica 怎么让 2000 条 WebSocket 不崩

本文是 Multica 拆解系列的第 7 篇。前六篇讲的是 Agent 的工位、大脑、生命周期、技能、组队、自治——每一篇都提到了一个东西但没展开:实时推送。你在网页上点"分配给 Claude Code",issue 状态秒变;daemon 跑完任务,进度条立刻动。中间发生了什么?这一篇拆到底。

这一篇要解决的问题

Multica 的实时性要求很极端:

你在网页上分配 issue,daemon 在你自己的 MacBook上领到任务——中间隔了一个 Go 服务器。任务状态变了,网页要在 100ms 内刷新。
你有 3 个前端:web(浏览器)、desktop(Electron)、mobile(React Native)。它们共用一套 WebSocket 协议,但 reconnect 策略完全不同。
服务器可能跑多个实例(水平扩展)。你在 instance A 上创建了一条 issue,但你的浏览器连着 instance B。B 怎么知道 A 上发生的事?

直觉上的做法是:一个 WebSocket Hub,所有连接挂上去,事件广播。这在单机没问题。但当你有多个服务器实例、上千个 WebSocket 连接、daemon 和浏览器各需要不同的推送粒度时,事情就复杂了。

我通读了 server/internal/realtime/ 和 server/internal/daemonws/ 的全部代码,找到了 Multica 的答案。核心是三层:双 Hub 分治、Dual-write 快路径、Redis Sharded Streams。

心智模型:两条 WS 通道 + 一个中继

架构图

三条规则先记住:

浏览器和 daemon 走两条不同的 WebSocket——不是一条复用。
浏览器 Hub 内部有 scope 房间(workspace / user / task / chat),不是全量广播。
跨实例的实时性靠 Redis Sharded Streams——不是 pub/sub,是 Stream + 消费组。

下面三个深度点逐层展开。

深度点一:为什么要两条 WebSocket?

Multica 有两个 WebSocket 端点,由两个完全独立的 Hub 管理:

浏览器 Hub(server/internal/realtime/hub.go,1020 行):

// server/internal/realtime/hub.go
type Hub struct {
    rooms      map[scopeKey]map[*Client]bool  // scope → 订阅者集合
    clients    map[*Client]bool               // 全部连接
    broadcast  chan []byte                     // 全局广播
    register   chan *Client                    // 注册/注销
    unregister chan *Client
    authorizer ScopeAuthorizer                 // scope 权限校验
}

连接时自动订阅两个 scope——workspace 和 user:

// server/internal/realtime/hub.go · Run()
case client := <-h.register:
    h.clients[client] = true
    // 自动订阅 workspace + user scope
    h.subscribe(client, ScopeWorkspace, client.workspaceID)
    h.subscribe(client, ScopeUser, client.userID)

之后客户端可以按需订阅更细的 scope——比如打开一条 issue 详情页时订阅 task:{taskID}:

// scope 类型常量
const (
    ScopeWorkspace = "workspace"
    ScopeUser      = "user"
    ScopeTask      = "task"
    ScopeChat      = "chat"
    ScopeDaemonRuntime = "daemon_runtime"
)

daemon Hub(server/internal/daemonws/hub.go):

// server/internal/daemonws/hub.go
type Hub struct {
    clients   map[*client]bool              // 全部 daemon 连接
    byRuntime map[string]map[*client]bool   // runtime ID → 连接
}

daemon 连上来之后按 runtime ID 索引,服务器派任务时可以按 runtime ID 精确推送到那台机器上的 daemon。

**为什么不复用一条 WebSocket?**因为两者的语义完全不同:

	浏览器 Hub	daemon Hub
认证	JWT / cookie	PAT (Personal Access Token)
推送粒度	scope 房间(workspace/user/task/chat)	runtime ID 精确投递
推送内容	业务事件(issue 状态变、聊天消息)	best-effort wakeup("有活可领了")
正确性保证	WS 是唯一来源	WS 只是提示,HTTP claim 才是正确性保证
心跳	ping/pong 54s	daemon heartbeat 15s

最后一行是关键设计:daemon 的 WebSocket 只负责"唤醒"——告诉 daemon "你有新任务可以来 HTTP 领了"。真正的任务领取走 HTTP ClaimTask。这样即使 WebSocket 断了,daemon 的 3 秒 HTTP 轮询也会兜底。这个决策让整个系统的容错性上了一个台阶:实时通道断不断,任务都不会丢。

深度点二:Dual-write 快路径 + Redis 8 分片

单实例时,事件从产生到推送很简单——直接调用 Hub 的 BroadcastToScope,走 in-memory channel,延迟 <1ms。

但多实例时,你在 instance A 创建了一条 issue,你的浏览器可能连着 instance B。B 的本地 Hub 里没有这个事件。怎么办?

直觉方案:Redis Pub/Sub

每个实例订阅 Redis 的 pub/sub channel,事件发到 Redis,所有实例收到后查本地 Hub 有没有订阅者。问题是:Pub/Sub 是 fire-and-forget。实例 B 刚好重启的那 2 秒,消息就丢了。

Multica 的方案:Redis Streams + 固定分片

Multica 用了 Redis Streams(XADD / XREAD BLOCK),不是 Pub/Sub。Stream 有持久化——消息写进去,即使消费者暂时不在,重连后也能从上次的位置继续读。

但 Stream 有一个问题:如果每个 scope 一条 Stream(比如 ws:scope:task:abc123:stream),那每个实例需要对每个活跃 scope 开一个 XREAD BLOCK 长连接。假设有 500 个活跃 task scope × 3 个实例 = 1500 个阻塞 Redis 连接。Redis 连接数是有限的。

Multica 的解法叫 Sharded Stream Relay(server/internal/realtime/sharded_stream_relay.go):

// server/internal/realtime/sharded_stream_relay.go
const (
    defaultShardedRelayShards       = 8          // 固定 8 个分片
    defaultShardedRelayStreamMaxLen = 100000     // 每 Stream 最多保留 10 万条
    defaultShardedRelayReadCount    = 128        // 每次最多读 128 条
    defaultShardedRelayReadBlock    = 5 * time.Second
)

所有事件写进 8 条固定的 Redis Stream(ws:relay:shard:0 到 ws:relay:shard:7)。写入时用 FNV hash 算分片号:

// ShardedStreamKey 返回分片对应的 Stream key
func ShardedStreamKey(shard int) string {
    return fmt.Sprintf("ws:relay:shard:%d", shard)
}

每个实例只需要 8 个 XREAD BLOCK 长连接——不管有多少个 scope。3 个实例 = 24 个阻塞连接,不是 1500 个。

读到消息后,本地过滤:看 Hub 里有没有人订阅了这个 scope,有就推送,没有就丢弃。这叫"读多过滤少"——宁可多读再扔,也不多连。

Dual-write:本地 + Redis 双写

事件产生时,不是只写 Redis 等其他实例读——那样本地实例的延迟也要等一次 Redis 往返。Multica 用了 Dual-write:

// 广播时同时走两条路
// 1. 本地 Hub 直接推送(fast path,<1ms)
hub.BroadcastToScope(scopeType, scopeID, message)

// 2. 写入 Redis Stream(给其他实例读)
relay.PublishWithID(scopeType, scopeID, "", message, ulid.Make().String())

本地浏览器的延迟是 in-memory 级别(<1ms),跨实例的延迟是一次 Redis 往返(~2-5ms)。

去重:128 条 LRU

Dual-write 有一个问题:其他实例从 Redis Stream 读到消息后,会把消息推给本地 Hub。如果"本地快路径"已经投递过了(同一个实例上的事件),客户端会收到两次。

Multica 给每个 WebSocket 连接挂了一个 128 条的 LRU 去重表,用 ULID 做 event ID:

// server/internal/realtime/hub.go
const dedupCapacity = 128

func (c *Client) markSeen(eventID string) bool {
    if eventID == "" { return true }
    c.dedupMu.Lock()
    defer c.dedupMu.Unlock()
    if _, ok := c.seenIDs[eventID]; ok {
        return false  // 重复,丢弃
    }
    c.seenIDs[eventID] = struct{}{}
    c.seenList = append(c.seenList, eventID)
    if len(c.seenList) > dedupCapacity {
        drop := c.seenList[0]
        c.seenList = c.seenList[1:]
        delete(c.seenIDs, drop)
    }
    return true
}

为什么不无限保留?因为内存。128 条足以覆盖 Dual-write 的去重窗口——Redis Stream 的消息从写入到被消费通常在毫秒级,远不会超过 128 条的范围。

深度点三:前端 reconnect 的三种策略

Multica 有三个前端,各自的 WebSocket reconnect 策略完全不同。我对比了它们的代码:

Web / Desktop:固定 3 秒

// packages/core/api/ws-client.ts
this.ws.onclose = () => {
    this.logger.warn("disconnected, reconnecting in 3s");
    this.reconnectTimer = setTimeout(() => this.connect(), 3000);
};

简单粗暴。断了就 3 秒后重连。为什么不用指数退避?因为 web/desktop 不存在"1000 个客户端同时断线"的场景——单用户的浏览器只有 1 个连接。

Mobile:指数退避 + Full Jitter

// apps/mobile/data/realtime/ws-client.ts
const RECONNECT_BASE_MS = 1_000;      // 基准 1 秒
const RECONNECT_CAP_MS = 30_000;      // 上限 30 秒
const RECONNECT_MAX_EXPONENT = 6;     // 最大指数 6 (1→2→4→8→16→32→cap 30s)

private nextDelay(): number {
    const exp = Math.min(this.reconnectAttempt, RECONNECT_MAX_EXPONENT);
    const raw = RECONNECT_BASE_MS * Math.pow(2, exp);
    const capped = Math.min(raw, RECONNECT_CAP_MS);
    // Full Jitter: 在 [0, capped] 范围内随机
    return Math.random() * capped;
}

为什么 mobile 要 jitter?因为 iOS 会在应用退到后台时杀掉所有 WebSocket 连接。当 1000 个用户同时打开 app 时,1000 个 WebSocket 同时断、3 秒后同时重连——服务器瞬间被"惊群"打爆。加了 jitter,每个客户端的 reconnect 时间在 0-30 秒之间随机散开,服务器压力被平滑掉。

三态生命周期:idle / active / paused

mobile 端还引入了一个 web 端没有的概念——三态生命周期:

// apps/mobile/data/realtime/ws-client.ts
type State = "idle" | "active" | "paused";

// idle    — 从未连接或已完全断开。onclose 不触发重连。
// active  — 想要连接。onclose → 安排重连。
// paused  — 调用者要求断开(app 退到后台)。onclose 不做事,resume() 重新打开。

iOS 会杀后台 WebSocket,如果不主动管理,退到后台再回到前台的瞬间,客户端会经历一次不必要的"断线 → 重连"抖动。三态模型让 provider 在 AppState = background 时调用 pause(),在 AppState = active 时调用 resume()——用户感知不到任何断线。

代码注释里写了一个 React Native 的经典坑:

// Detaches socket handlers BEFORE close() to avoid the spurious
// onclose-triggered reconnect that bites RN (close is async over
// the bridge — see facebook/react-native#9465).

React Native 的 WebSocket.close() 是异步过 bridge 的——调用后 onclose 不是立刻触发,而是延迟到 bridge 下一次 flush。如果不提前 detach handler,onclose 会在 pause() 之后错误地触发一次重连。这个 bug 在 web 端不存在,因为浏览器的 close() 是同步的。

认证:Token 不进 URL

一个安全细节。浏览器 WebSocket 不能设置自定义 Header,很多应用把 token 放在 URL query parameter 里(wss://server/ws?token=xxx)。但 URL 会被代理日志、CDN 缓存、浏览器历史记录捕获。

Multica 的做法是:token 作为 WebSocket 连接建立后的第一条消息发送:

// packages/core/api/ws-client.ts
this.ws.onopen = () => {
    if (!this.cookieAuth && this.token) {
        this.ws.send(JSON.stringify({ type: "auth", payload: { token: this.token } }));
        return;
    }
    this.onAuthenticated();
};

服务器收到 auth 消息后才认为连接已认证,之前不推送任何业务事件。认证成功后回复 auth_ack。

全栈联调闭环:一次 issue 状态变更的完整旅程

把前六篇和这一篇的拼图合到一起,一次"分配 issue 给 Claude Code"的全栈实时闭环:

你在网页上点"分配给 Claude Code"
前端发 HTTP POST /issues/:id/assign,乐观更新 issue 卡片状态
Go 服务器写入 DB,产生 task:dispatched 事件
事件进入 Dual-write:本地 Hub 推给你的浏览器(web 实时刷新,你看到状态秒变);同时 XADD 进 Redis Sharded Stream
其他实例从 Redis Stream 读到事件,推给它们上面的浏览器(如果你开了第二个 tab)
同一条事件通过 daemon_runtime scope 推给 daemon Hub
daemon Hub 按 runtime ID 找到你的 MacBook 上的 daemon 连接,发一条 best-effort wakeup
daemon 收到 wakeup(或者 3 秒 HTTP 轮询到了),调 ClaimTask 领取任务
daemon 启动 Claude Code,开始写代码
每个 token 输出通过 daemon WebSocket 流回服务器,再通过浏览器 Hub 推到你的网页——你看到实时进度