群聊实现与扩散策略落地

群聊的技术难点从来不是「能不能发出去」，而是「一条消息要不要变成 N 条」。500 人的群如果照单聊的做法给每个成员都写一份、投一次，单条消息的写入放大就是 500 倍，一次热群刷屏就能把数据库写满。本章按成员规模分档：小群走写扩散（write fanout），大群走读扩散（read fanout），并给出两者的落地代码与存储落点。

本章目标

把群 group_id 接进第 07 章的单聊链路：SendMessage 收到 conv_type=2 时走群流程，成员数 ≤ 200 的群为每个成员写一份收件箱并投递，成员数更大的群只落一份群消息、由成员按游标增量拉取。

两种扩散方式

维度	写扩散（小群，≤ 200 人）	读扩散（大群，> 200 人）
落库代价	O(成员数)：每个成员一行收件箱	O(1)：群消息只在 `msg` 留一份
离线留存	成员各自的收件箱，确认后回收	群会话日志本身，按位点增量拉
拉取路径	`offline_msg_xx`（按 uid 分片）	`msg_xx`（按 conv_id 分片）+ `user_sync_cursor`
主要风险	成员接近上限时插入变慢	位点丢失导致重放或漏拉
典型场景	家庭群、项目群	万人直播群、通知群

两种方式在在线投递这一层完全一样，只是「离线留存」的存法不同，因此可以共用一套投递代码，只在落库与拉取处按规模分岔。

复用的四张表

msg_00 ~ msg_15        会话日志，按 conv_id % 16 分片（第 11 章），群消息只落一份
offline_msg_00 ~ 15    收件箱，按 uid % 16 分片，主键 (uid, conv_id, seq)（第 09 章）
user_sync_cursor       设备拉取位点，主键 (uid, device_id, conv_id)（第 09 章）
conv_member            群成员表，主键 (conv_id, uid) + KEY (uid, join_at)（第 11 章）

群聊不需要新的存储结构：写扩散就是把 offline_msg 多写几行，读扩散就是完全不写收件箱。

分档入口

const fanoutThreshold = 200 // 成员数 ≤ 200 走写扩散，> 200 走读扩散

// Send 群消息入口：无论哪种扩散，都先把群消息落进会话日志拿到 seq，
// 再按成员规模决定扩散方式——两种方式对客户端保持一致的可见语义。
func (s *GroupService) Send(ctx context.Context, in GroupSendInput) (*SendResult, error) {
	if err := s.groups.CheckMember(ctx, in.GroupID, in.FromUID); err != nil {
		return nil, err // ErrNotMember：非群成员直接拒绝，不能靠客户端自觉
	}
	// 每个群对应一个会话：conv_id 在建群时生成，成员关系记在 conv_member（第 11 章）
	convID, err := s.convs.ByGroup(ctx, in.GroupID)
	if err != nil {
		return nil, fmt.Errorf("load group conv group=%d: %w", in.GroupID, err)
	}
	msg, err := s.msgs.Insert(ctx, convID, in.FromUID, in.Payload) // 事务内 conv_seq 取号
	if err != nil {
		return nil, fmt.Errorf("insert group message conv=%d: %w", convID, err)
	}
	members, err := s.groups.MemberCount(ctx, convID)
	if err != nil {
		return nil, fmt.Errorf("count members conv=%d: %w", convID, err)
	}
	if members <= fanoutThreshold {
		if err := s.fanoutWrite(ctx, msg); err != nil {
			// 写扩散失败不回滚：群消息已落库，成员下次拉取会按位点补齐
			slog.Error("write fanout failed", "conv_id", convID, "msg_id", msg.MsgID, "err", err)
		}
	} else {
		slog.Info("large group, read fanout", "conv_id", convID, "members", members)
	}
	return &SendResult{MsgID: msg.MsgID, Seq: msg.Seq, SentAt: msg.CreatedAt}, nil
}

小群写扩散

// fanoutWrite 小群写扩散：为每个成员写一行收件箱，并给在线成员逐端投递。
func (s *GroupService) fanoutWrite(ctx context.Context, msg *model.Message) error {
	members, err := s.groups.MemberIDs(ctx, msg.ConvID, fanoutThreshold)
	if err != nil {
		return fmt.Errorf("load members conv=%d: %w", msg.ConvID, err)
	}
	if err := s.inbox.BatchInsert(ctx, members, msg); err != nil { // 一次多值 INSERT，别逐条插
		return fmt.Errorf("batch insert inbox conv=%d: %w", msg.ConvID, err)
	}
	routes, err := s.routes.LookupBatch(ctx, members) // pipeline：N 次 HGETALL 压成一次往返
	if err != nil {
		return fmt.Errorf("lookup batch routes conv=%d: %w", msg.ConvID, err)
	}
	for uid, byDevice := range routes { // 只投在线成员，离线成员靠收件箱留痕
		for _, rt := range byDevice {
			if err := s.deliver.Push(ctx, rt, msg.AsPushFrame()); err != nil {
				slog.Warn("group deliver failed", "uid", uid, "conn_id", rt.ConnID, "err", err)
			}
		}
	}
	return nil
}

大群一侧则是「落库即结束」，不再扇出：成员上线或进入会话时调 PullGroup(ctx, uid, convID, sinceSeq, limit) 从 msg_xx 按会话 seq 增量拉取（limit 钳到 500 以内），客户端确认收下后再推进 user_sync_cursor——UPDATE user_sync_cursor SET ack_seq = GREATEST(ack_seq, ?) WHERE uid = ? AND device_id = ? AND conv_id = ?，GREATEST 保证位点只增不减，补拉旧消息不会把进度打回去（第 09 章）。

验证方式

# 造一个小群（conv_id=88001，5 人）和一个大群（conv_id=88000，500 人）
mysql -uroot -p im -e "INSERT INTO conv_member (conv_id, uid, join_at) SELECT 88000, uid, NOW() FROM user LIMIT 500;"
# 小群发一条：期望收件箱新增 5 行（每个成员一行），消息落在 msg_01（88001 % 16）
grpcurl -plaintext -d '{"client_msg_id":"g-1","from_uid":"1001","group_id":"1001",
  "conv_type":2,"payload":"aGk="}' 127.0.0.1:9090 im.v1.LogicService/SendMessage
mysql -uroot -p im -e "SELECT COUNT(*) FROM offline_msg_09 WHERE uid=1001 AND conv_id=88001;" # 期望：1（1001 % 16 = 9）
# 大群发一条：日志出现 large group, read fanout；收件箱不新增，消息只在 msg_00
mysql -uroot -p im -e "SELECT COUNT(*) FROM offline_msg_09 WHERE uid=1001 AND conv_id=88000;" # 期望：0
mysql -uroot -p im -e "SELECT seq,from_uid FROM msg_00 WHERE conv_id=88000 ORDER BY seq DESC LIMIT 1;"

常见坑

坑	现象	正确做法
阈值拍脑袋写死	成员接近上限的群忽然变慢	用「成员数 × 单条插入耗时」实测后定档，并单独压测临界群
逐条 `INSERT` 扇出	200 人就是 200 次网络往返	一次多值 INSERT，路由用 pipeline 批量查
大群一次查出全部成员	内存与 GC 抖动，OOM	`MemberIDs` 必须带 limit 与游标分批
只靠 Redis 缓存群成员	退群的人还一直收到消息	缓存带版本号或短 TTL，写路径以 DB 为准
拉取时全量重放	客户端一进群就卡死	必须带 `sinceSeq` 与 limit，位点丢失才按时间回捞尾页
退群后位点未清理	重新入群时从旧位点续拉，漏掉中间消息	退群删 `user_sync_cursor` 对应行，重新入群从当前 seq 开始

小结：群聊按成员规模分档——小群写扩散，往 offline_msg 多写成员数行收件箱，换读路径的一次单分片扫描；大群读扩散，消息只在 msg_xx 留一份，成员按 user_sync_cursor 只增不减的位点增量拉取；两种方式共用同一套在线投递代码，对客户端保持一致的可见语义。

笔记加载中…

标签：#IM #Go #项目实战

返回知识库