后端系统工程 08：群消息扇出与成员版本

本批导航：00 技术方案怎样接受验证｜01 API 契约与业务不变量｜02 负载模型与容量预算｜03 单库基线与故障实验工具｜04 持久化边界｜05 长连接接入｜06 发送确认｜07 排序与增量同步｜08 群消息扇出｜09 多设备已读｜10 网关故障与路由收敛｜11 撤回与集成验收

单聊的一条消息只送达两个人的几台设备，群消息要送达几十到几百名成员的设备。第 02 篇的模型已经表明，扇出带来的送达成本远大于消息写入本身，而且随群规模线性增长。本篇处理群消息的两个问题：送达成本怎样控制，以及在成员不断加入、退出的情况下，每条消息应该送给谁、谁可以读取它。

第二个问题比第一个更容易出错。成员变更与消息发送是并发的；推送是异步的，从消息提交到推送发出之间，成员可能已经变化。若推送目标按“推送时刻的成员”计算，刚退群的成员可能收到退群之后的消息，刚入群的成员可能收到入群之前的消息。本篇按事件序号计算成员资格，使这两种情况都不会发生。

1. 需求与不变量

编号	不变量	对应需求
F1	序号为 s 的事件，只推送给在 s 上属于成员的用户	R3
F2	用户只能通过同步或单条读取获得其成员区间内的事件；权限在读取时判定	R3
F3	新成员看不到入群之前的事件；退群成员保留区间内的历史，包括自己的退群事件，看不到之后的事件	R3
F4	退群后再入群，两个区间之间的事件保持不可见	R3
F5	两种投递模式下，在线且持续阅读的设备最终都持有全部可见事件	R5
F6	投递模式是会话属性，可以针对单个会话切换，不影响同步协议	—

“成员版本”在本实现中不是一个单独的版本号，而是会话序号本身：成员变更是会话日志中的事件（第 04 篇的 P4），它们和消息共享序号。“序号 s 上的成员集合”就是该会话的成员版本。这避免了另外维护一个成员版本号并与消息序号对齐的问题。

2. 规模假设

第 02 篇的基线：群消息占 40%，平均群规模 30 人，每用户 1.6 台在线设备，发送时成员在线率 0.35。每条群消息约送达 (30 × 1.6 − 1) × 0.35 ≈ 16.45 台在线设备。群规模超过 200 人时切换为提示模式，收到提示后立即拉取的设备比例假设为 20%。

本篇的本地实验使用 40 名成员、30 台在线设备、20 条消息，每条正文 200 个字符。这个规模只用来比较两种模式的机制差异，不用于推算容量。

3. 两种投递模式

模式	网关下发	设备动作	数据库读取	送达确认
push	完整事件 event{ev}	直接应用，确认	每条事件读一次，随后对每个目标连接复用	每台在线设备确认
pull	提示 hint{conv, head}	更新本地已知 head（显示未读）；打开会话时才同步	只有实际同步的设备产生读取	只有同步的设备确认

图 8-1  同一条群消息在两种模式下的下发路径
                         ┌── push 模式 ───────────────────────────────────────┐
提交 → NOTIFY → 网关 ──→ │ fanout(conv, seq): 读事件 + 读 seq 上的成员          │
                         │ 对每个在线成员设备: event{完整事件} (约 400 B)        │
                         │ 设备: 应用 → 持久化 → ack                            │
                         └────────────────────────────────────────────────────┘
                         ┌── pull 模式 ───────────────────────────────────────┐
                         │ fanout(conv, seq): 只读 seq 上的成员                  │
                         │ 对每个在线成员设备: hint{conv, head} (约 40 B)        │
                         │ 设备: knownHead = max(knownHead, head)               │
                         │   打开会话的设备: 合并提示 → sync(after=cursor) → ack │
                         │   未打开的设备: 只显示未读数，不拉取                  │
                         └────────────────────────────────────────────────────┘

两种模式共用第 07 篇的同步协议，客户端的游标、缺口检测和本地存储逻辑完全相同。模式只决定“通知里带不带内容”。因此切换模式不需要迁移数据，也不需要客户端升级，只需修改会话的 delivery_mode 字段（F6）。

4. 接口与数据模型

本篇没有新增接口。用到的是第 01 篇已定义的 event、hint、sync、fetch。

成员资格使用第 04 篇的 membership_periods：

CREATE TABLE membership_periods (
    conv_id  text NOT NULL REFERENCES conversations,
    user_id  text NOT NULL REFERENCES users,
    from_seq bigint NOT NULL,
    to_seq   bigint,
    PRIMARY KEY (conv_id, user_id, from_seq),
    CHECK (to_seq IS NULL OR to_seq >= from_seq)
);
CREATE UNIQUE INDEX one_open_period ON membership_periods (conv_id, user_id) WHERE to_seq IS NULL;

序号 s 上的成员：from_seq <= s AND (to_seq IS NULL OR s <= to_seq)。

操作	在锁下执行的动作	区间变化
入群	分配序号 j，写 join 事件	新增 [j, NULL)
退群	分配序号 l，写 leave 事件	开放区间改为 [from, l]
再入群	分配序号 r，写 join 事件	新增 [r, NULL)

5. 正常路径与关键实现

5.1 推送目标按事件序号计算

// examples/im/src/im/Store.java（节选）
public Fanout fanout(String convId, long seq) {
    return db.tx(c -> {
        String mode = …;   // SELECT delivery_mode FROM conversations WHERE conv_id=?
        List<String> users = …;
        // SELECT DISTINCT user_id FROM membership_periods
        //  WHERE conv_id=? AND from_seq<=? AND (to_seq IS NULL OR to_seq>=?)   -- 参数均为 seq
        Map<String, Object> event = mode.equals("push") ? loadEventOrNull(c, convId, seq) : null;
        return new Fanout(event, users, mode);
    });
}

网关收到通知 conv:seq 时调用 fanout，对返回的每个用户查找本网关上的连接，下发事件或提示（Gateway.onEvent）。

查询条件中的参数是事件的序号，而不是“当前”。设想这样一个交错：

1. 成员 u-001 在线。
2. 事务 A 写入消息，得到序号 69，提交。
3. 事务 B 让 u-001 退群，得到序号 70，提交。
4. 网关处理消息 69 的通知。

此时按“当前成员”计算，u-001 已经不在群里，不会收到消息 69，但 69 本来属于 u-001 的可见范围，它可以通过同步拿到，只是推送缺失了，这不破坏正确性。反过来的交错才是危险的：

1. 事务 B 让 u-001 退群，得到序号 68，提交。
2. 网关因为某种原因（监听线程繁忙）还没有更新任何内存中的成员缓存。
3. 事务 A 写入消息，得到序号 69，提交。
4. 网关处理消息 69 的通知，按缓存中的成员推送，u-001 收到了退群之后的消息。

按序号计算时，序号 69 上 u-001 的区间是 [1, 68]，不包含 69，u-001 不在目标之列。成员检查与成员变更共用会话行锁分配序号，所以“69 在 68 之后”是确定的。本实现没有在网关缓存成员列表，每次扇出都查询数据库，代价是每条群消息一次额外查询；引入缓存时，缓存必须以序号作为版本，过期判定为“缓存版本 < 事件序号”。

5.2 读取时判定

推送只是一个副本。退群的成员可能在退群之前已经知道后续消息的序号（例如收到了提示），可以尝试直接用 fetch 读取，也可以用旧游标同步。这些读取都在执行时检查成员区间：

// examples/im/src/im/Store.java（节选）
public Map<String, Object> eventFor(Principal who, String convId, long seq) {
    return db.tx(c -> {
        if (!visible(c, convId, who.userId(), seq))   // from_seq <= seq <= coalesce(to_seq, ∞)
            throw new ImException("NOT_MEMBER", "event not visible");
        return loadEvent(c, convId, seq);
    });
}

同步查询对每个事件用 EXISTS 子查询过滤（第 07 篇 5.3 节），所以多个区间（F4）自然得到正确结果。

5.3 新成员的游标

新成员第一次同步时 after=0。服务端只返回入群事件及之后的事件，next_after=head。如果 next_after 取“最后返回的序号”，在没有更多事件时也成立；但在退群后的同步中，next_after 必须越过不可见的尾部，这就是第 07 篇规定 has_more=false 时 next_after=max(after, head) 的原因。

5.4 提示的合并

提示模式下，一个设备可能在短时间内收到大量提示。客户端对每个会话最多保持一个进行中的同步，期间到达的提示只设置一个“再同步一轮”的标记（第 07 篇 5.2 节）。未打开会话的设备只更新 knownHead，用于显示未读数，不发起同步。

6. 方案比较

6.1 扇出在哪里发生

方案	写入	读取	成员变更
写扩散：每个成员一份收件箱	每条消息写 N 行	按用户读收件箱，一次查询得到所有会话的新消息	入群时需要回填或接受“入群前无历史”；退群时需要清理
读扩散：每个会话一份日志	每条消息写 1 行	按会话读日志；查询“所有会话的新消息”需要逐会话比较游标	只修改成员区间
混合：小群写扩散，大群读扩散	介于二者之间	两套读取路径	两套变更逻辑

本系列采用读扩散，理由在第 02 篇第 6 节给出：在本组假设下数据库写入已是瓶颈，写扩散放大约 12 倍。读扩散把扇出推迟到送达环节，送达环节再用两种模式控制成本。

6.2 推送还是提示

因素	偏向推送	偏向提示
设备的阅读比例	高（工作群、多数设备正在看）	低（大群、多数设备只看未读数）
群规模	小	大
首条可见延迟	推送少一次往返	多一次同步往返
数据库读取	每条事件一次读取，每台设备一次确认写入	只有阅读的设备读取与确认
网关帧大小	大（完整事件）	小（约 40 字节）

这张表说明不存在通用的优胜者。第 02 篇以 200 人为阈值，是对“阅读比例随群规模下降”的假设；本篇的本地实验从另一个角度展示了同样的结论。

6.3 成员资格的表示

第 04 篇比较了当前成员表、加入时间与序号区间三种表示。本篇的竞态分析补充了一个理由：只有序号区间能让推送目标和读取权限按同一个依据计算。若推送用缓存的当前成员，读取用加入时间，两者在边界上会给出不同的答案，出现“推送收到了但同步读不到”或者相反的情况。

7. 验证

运行方式：真实 PostgreSQL 16、单个网关 JVM（IM_OUTBOX=256，避免小队列在这组负载下溢出影响字节统计）、测试进程内的 30 个客户端。字节数在客户端统计，为 JSON 文本长度，不含 WebSocket 帧头。结果见 evidence/im-08/result.json。

7.1 两种模式的对比（观测，不判优劣）

同一组成员，三个会话分别为推送模式、提示模式（全部设备阅读）、提示模式（只有 1/5 设备阅读）。每个会话发送 20 条消息。

会话	读者设备	下发帧数	同步次数	下发帧字节	同步回复字节	服务端到客户端合计
g-push	30	600 个事件	0	241 500	0	241 500
g-pull	30	600 个提示	145	22 800	250 938	273 738
g-lazy	6	600 个提示	56	22 800	53 966	76 766

读数要点：

• 600 = 30 台设备 × 20 条消息。三种情况下帧数相同，差别在于帧的内容：推送帧平均约 402 字节，提示帧约 38 字节。
• 全员阅读时，提示模式的总字节数（273 738）比推送模式（241 500）多约 13%。145 次同步没有被合并成 30 次：提示到达的节奏与同步往返交错，每台设备平均同步约 4.8 轮，每轮都要付出回复的固定开销（head、next_after 等字段）。同步次数取决于调度，归档的上一次运行是 72 次，断言没有对它做要求。
• 只有 1/5 设备阅读时，提示模式的字节数约为推送模式的 32%。另外 24 台设备只收到提示、更新了已知 head，没有拉取正文。

检查 08-both-modes-complete 断言三个会话中所有读者设备最终持有全部消息，所有设备都已知 head（F5），结果 PASS。08-amplification-compared 与 08-pull-lazy-readers 是观测类检查，只记录数值。

7.2 成员区间

类型	检查	预期	观测
边界	08-join-no-prior-history	u-new 在序号 62 入群；同步只从 62 开始，看不到入群前的消息；next_after 到达 head	首个可见 62，未见入群前消息，next_after=63
边界	08-leave-closes-interval	退群序号 64；能看到自己的 leave 事件，看不到之后的消息；退群后发送得到 NOT_MEMBER	均符合
边界	08-rejoin-gap-stays-hidden	序号 66 再入群；64 与 66 之间的消息不可见，66 之后的可见	均符合
失败	08-no-push-after-leave	在线设备 u-001 订阅推送后退群（序号 68），随即写入消息 69；1 秒内该设备未收到 69；直接 fetch 69 得到 NOT_MEMBER	未泄露，NOT_MEMBER

最后一项验证了 5.1 节的竞态处理：消息 69 在退群 68 之后提交，按序号计算的推送目标不包含 u-001。这项检查的局限是：它验证的是“退群在前、消息在后”的顺序，没有逐一枚举网关处理通知时的所有交错；正确性的论证依赖于查询条件本身，而不依赖于时序。

8. 故障与恢复

故障	影响	恢复
推送模式下某设备的推送丢失	该设备缺少事件	缺口检测或重投扫描（第 06、07 篇）
提示丢失	设备的已知 head 偏小，未读数偏小	下一个提示或扫描提示；打开会话时同步
大群推送导致某连接待发队列溢出	该连接以 4008 关闭	重连后同步（第 05 篇）；大群应改用提示模式
成员变更与发送并发	—	由会话行锁确定先后，推送目标按序号计算
网关处理通知时数据库不可用	fanout 失败，该事件不推送	重投扫描在数据库恢复后补发提示
退群用户持有旧游标继续同步	—	只返回到退群序号，next_after 越过其后的范围

大群在推送模式下的待发队列溢出值得单独说明。一个 500 人的群在一秒内收到 20 条消息时，每台在线设备的待发队列要容纳 20 帧，每帧数百字节；如果同时还有其他会话的推送，默认 64 帧的上限可能被突破。这是第 02 篇设置阈值的另一个原因：提示模式不仅减少字节，也减少了单个连接的瞬时帧数，因为客户端的提示合并让多个提示只引发少量同步。

9. 限制

• 每次扇出都查询数据库中的成员，没有缓存。大群在高频写入时，这个查询本身会成为负担；带版本的缓存未实现。
• 提示模式下按“每个提示 40 字节”的下发仍然是每台在线设备一帧。第 02 篇已指出，1000 人群时帧数本身会超出网关预算，需要按时间窗口合并提示或只向打开会话的设备发提示，本批未实现。
• 两种模式的字节对比只覆盖应用层 JSON，不含帧头、TLS 和 TCP 开销；同步次数依赖调度。
• 实验在单网关上进行；多网关时每个网关各自查询一次成员，查询次数随网关数增长，这一点未测量。

下一篇处理同一用户的多台设备：送达状态按设备记录，已读状态按用户记录，二者怎样互不干扰，未读数怎样从它们推导出来。

参考资料

• PostgreSQL 16 文档，9.23 Subquery Expressions（EXISTS）。https://www.postgresql.org/docs/16/functions-subquery.html
• PostgreSQL 16 文档，13.3.2 Row-Level Locks（成员变更与发送共用会话行锁）。https://www.postgresql.org/docs/16/explicit-locking.html
• PostgreSQL 16 文档，NOTIFY（通知按提交顺序投递，载荷长度限制）。https://www.postgresql.org/docs/16/sql-notify.html