聊天记录分桶存进 MongoDB：一条消息占 1/100 个文档，位置全靠算

一条聊天消息在库里不是一行，也不是一个文档，而是 1/100 个文档。

const singleGocMsgNum = 100 // pkg/common/storage/model/msg.go:11

一个 MongoDB 文档里用数组装 100 条消息，第 101 条进下一个文档。这就是分桶（Bucket Pattern）。下面按「为什么分桶 → 怎么寻址 → 怎么幂等写 → 怎么读 → 读时怎么改写」走一遍。

源码位置（同一份仓库）：https://github.com/CYQ111999/IM-Server，基线是 OpenIM Server v3.8 的瘦身版，main 分支。本篇涉及 pkg/common/storage/model/msg.go、pkg/common/storage/database/mgo/msg.go、pkg/common/storage/controller/msg.go。

一条消息一个文档，问题出在哪

第三点最容易被忽略。MongoDB 的索引是 B 树，100 万个小文档的索引，比 1 万个「每桶 100 条」的文档索引大得多，而索引常驻内存。

分桶本质是拿「一次 IO 取回 100 条」来摊薄这些开销。和 InnoDB 的页（16KB 一页装多行）是同一个思路：存储系统的性能，取决于一次 IO 能拿回多少有用数据。

// model/msg.go:20-23 一个 MongoDB 文档
type MsgDocModel struct {
DocID string          `bson:"doc_id"`
Msg   []*MsgInfoModel `bson:"msgs"` // 长度固定 100，空洞是 nil
}

寻址：四个纯函数决定一切

分桶之后，「这条消息在哪个文档、哪个下标」必须能直接算出来，不能查。

// model/msg.go:94-127（删去无关行）
func (m *MsgDocModel) GetDocIndex(seq int64) int64 { return (seq - 1) / singleGocMsgNum }
func (m *MsgDocModel) GetDocID(conversationID string, seq int64) string {
return m.indexGen(conversationID, (seq-1)/singleGocMsgNum)
}
func (m *MsgDocModel) GetMsgIndex(seq int64) int64 { return (seq - 1) % singleGocMsgNum }
func (m *MsgDocModel) indexGen(conversationID string, seqSuffix int64) string {
return conversationID + ":" + strconv.FormatInt(seqSuffix, 10)
}

(seq-1) 的减一不能省。写成 seq/100 会出两种错：

• seq=100 算成 docIndex 1、arrayIndex 0，和 seq=101 撞到同一个位置；
• seq=0（不存在）算成 docIndex 0，把空洞当成有效消息。

还有一个隐含契约：数组永远 100 个槽，空洞用空对象占位。破坏长度，(seq-1)%100 算出来的下标全部错位。

写入：先 update 后 insert 的幂等状态机

Kafka 是 at-least-once，同一批消息可能被消费两次，所以写入必须幂等：写两次和写一次结果一样。

这里不能靠唯一索引兜底——消息在数组里，MongoDB 没法给「数组元素的某个字段」建唯一索引来防重复插入同一个位置。于是走「先 update，不成再 insert」的状态机（controller/msg.go:120 附近）：

tryUpdate := true // 乐观假设：目标文档已存在
for i := 0; i < len(fields); i++ {
seq := firstSeq + int64(i)
if tryUpdate {
matched := UpdateMsg(docID, index, "msg", field) // $set msgs..msg
if matched {
continue // 位置已有数据 → 幂等命中
}
}
// 位置是空的 → 构造整个文档，把「这一段」全部塞进去，一次插入
doc := MsgDocModel{DocID: ..., Msg: make([]*MsgInfoModel, 100)}
insert := 0
for j := i; j < len(fields); j++ {
if GetDocID(conversationID, firstSeq+int64(j)) != doc.DocID {
break // 跨桶了，剩下的留给下一轮
}
doc.Msg[GetMsgIndex(firstSeq+int64(j))] = &MsgInfoModel{Msg: fields[j]}
insert++
}
fillNil(&doc) // 空洞补 &MsgInfoModel{}，别留 null
if err := Create(ctx, &doc); err != nil {
if mongo.IsDuplicateKeyError(err) { // 别人抢先插了
i--              // 退回这个位置
tryUpdate = true // 后面全改用 update
continue
}
return err
}
tryUpdate = false // 已经插过一块，后面继续 insert
i += insert - 1   // 跳过刚插进去的这一段
}

$set msgs..msg 命中（matched）就说明这个位置已经有消息，直接跳过；位置为空才构整个文档一次插入；插入撞 DuplicateKeyError 说明并发下别人先建了同一个 docID，退回该位置改走 update。三条分支合起来，重复消费落到同一位置是覆盖而不是追加。

读：缓存 → 文档 → 下标

以拉 seq 10001~10020 为例，分三层。

第一层，按 seq 单条缓存（Redis，TTL 24 小时）。GetMessageBySeqs 把 seq 列表拆成命中与未命中，未命中的聚合成一次回源。为什么不按文档缓存？因为撤回、删除、已读这些操作都只影响一条消息；按文档缓存的话，撤回一条就要失效 100 条消息的缓存。

第二层，按 docID 分组，一次读一个文档：

func (m *MsgDocModel) GetDocIDSeqsMap(conversationID string, seqs []int64) map[string][]int64 {
t := make(map[string][]int64)
for _, seq := range seqs {
docID := m.GetDocID(conversationID, seq)
t[docID] = append(t[docID], seq)
}
return t
}

20 条连续消息通常落在一个 docID 里，也就是 1 次 Mongo 查询——读放大从 20 次变成 1 次。

第三层，从数组里取下标，GetMsgIndex(seq) 直接定位。

读时改写：存一份事实，按当前用户投影

同一条群消息，发送者 A、已读的 B、删掉这条的 C、以及撤回之后的所有人，看到的样子都不一样。如果每个视角存一份，一条群消息 × 500 人 = 500 份，写放大直接不可接受。

所以是读时改写（read-time projection）：存一份事实 + 极少量「谁对它做了什么」的标签，读的时候用「事实 + 标签 + 当前用户」现场算出视图。标签只有三类：

读的时候三步：拿原文 → 判断 userID ∈ del_list → 判断 revoke != nil，该清空的清空，该换文案的换文案。群消息在库里始终只有一份。

取舍与限制

100 这个数字是权衡，不是推导：单文档 16MB 上限决定了桶不能太大；每次更新都会整文档重写，桶越大写放大越明显；索引项数量则限制了桶不能太小。真要换别的值，四个寻址函数里只有一个常量，但历史数据的 docID 全部作废，等于要做一次数据迁移。

这套设计不适用的场景也很明确：

• 需要跨桶的范围查询（比如按时间扫全会话），分桶反而增加遍历成本；
• 需要按消息内容检索，数组里的元素在索引上并不友好；
• 消息总量极少但会话极多，桶长期半空，等于用大文档换了个寂寞。

最后交代信源：这个仓库是在实习项目基础上改造的，star 很少，没有做过生产验证，上面的写入并发、桶内空洞比例对实际性能的影响都没有实测数据。接入前建议自己按目标量级压一轮写入和拉历史。

小结：100 条一个文档，(seq-1)/100 定文档、(seq-1)%100 定下标；del_list / revoke / is_read 三个标签撑起「每个人看到的不一样」；一条契约——数组永远 100 个槽，破坏它寻址全错。