单聊消息全链路实现

前面四章分别做出了网关、协议、鉴权和逻辑层，但「A 发一句话，B 立刻收到」这条链路还没有一次真正跑通。本章把上行接入、鉴权、入库、路由查询、投递、ACK 与回执七跳串起来，每一跳都写明成功语义与失败后的责任方——这是 IM 项目里最容易「看起来能跑、线上一堆丢消息」的地方。

本章目标

用两个 WebSocket 客户端跑通单聊：A 发出 send_msg 帧后收到带 msg_id、seq 的响应，B 收到 push 帧并回 ack 后 A 收到回执，而 MySQL 里只有一条消息记录。

全链路时序

A ─①send_msg→ G1 ─②SendMessage(gRPC)→ L ─③鉴权 + 幂等校验（client_msg_id）
                                        ├─④写 MySQL：msg 分片 + conv_seq 取号（同一事务）
                                        └─⑤查路由：Hub 本机连接 → Redis route:uid:{uid}
L ─⑥响应信封→ G1 → A               （代表「服务端已收妥」，此刻消息已落库）
L ─⑦Deliver→ G2（远端网关）或本机直投（同一台机器时省掉一跳）
G2 ─⑧push 帧→ B ─⑨ack→ G2 ─⑩AckMessage(gRPC)→ L ─⑪回执信封→ A

四段各自的成功语义必须分清：上行成功 = 服务端收妥（可安全清掉本地重发队列），投递成功 = 对端网关收下，push 帧到达 = 对端设备收到，回执 = 对方用户端确认。任何一段失败都只回退这一段，绝不能因为投递失败就把已落库的消息删掉或回滚。

上行：网关解帧后调 gRPC

网关这一侧只做四件事：用第 03 章的 Envelope 解出 send_msg，带上 3 秒超时和 x-trace-id 调用 LogicService.SendMessage，成功把 msg_id/seq 填回 code=0 的响应信封，失败回非 0 错误码并标出可否重发。只有 Internal 与 Unavailable 两个状态码才让客户端重发，InvalidArgument、PermissionDenied 重发也是白重发。鉴权在连接建立时已经完成（第 04 章），这里只信任会话里的 uid，绝不解析客户端报文里自带的 from——那是伪造身份的经典入口。

逻辑层：落库与投递编排

// Send 是单聊的核心编排：拿会话 → 分配 msg_id → 幂等落库 → 查路由投递。
// 落库成功即代表「服务端已收妥」；投递失败只影响送达，不影响消息的存在性。
func (s *MessageService) Send(ctx context.Context, in SendInput) (*SendResult, error) {
	// 单聊也走会话表：conv 分配 conv_id、conv_member 写两条成员行（第 11 章），
	// 这样群聊的扩散、权限、已读逻辑与单聊共用一套代码。
	convID, err := s.convs.EnsureSingle(ctx, in.FromUID, in.ToUID)
	if err != nil {
		return nil, fmt.Errorf("ensure single conv %d->%d: %w", in.FromUID, in.ToUID, err)
	}
	msg := &model.Message{
		MsgID: s.idgen.Next(), ClientMsgID: in.ClientMsgID, // snowflake ID + 幂等键（第 10 章）
		ConvID: convID, FromUID: in.FromUID, ToUID: in.ToUID,
		Payload: in.Payload, CreatedAt: time.Now().UnixMilli(),
	}
	seq, err := s.msgs.Insert(ctx, msg) // 事务：conv_seq 取号 + 插 msg 分片 + 写收件人收件箱
	if err != nil {
		if store.IsDuplicate(err) { // 唯一索引冲突说明是重发，返回已存在的那条
			return s.msgs.ByClientMsgID(ctx, in.FromUID, in.ClientMsgID)
		}
		return nil, fmt.Errorf("insert message: %w", err)
	}
	if err := s.deliver.ToUser(ctx, in.ToUID, msg); err != nil {
		// 投递失败不回滚落库：消息已在库里，收件人上线后由 PullOffline 补齐
		slog.Warn("deliver failed, fallback offline", "msg_id", msg.MsgID, "err", err)
	}
	return &SendResult{MsgID: msg.MsgID, Seq: seq, SentAt: msg.CreatedAt}, nil
}

投递：多端扇出与失败摘路由

// ToUser 把一条已落库的消息投给目标用户的全部在线设备（多端同时投）。
func (d *Deliverer) ToUser(ctx context.Context, uid uint64, msg *model.Message) error {
	routes, err := d.routes.LookupAll(ctx, uid) // 一次 HGETALL 取回全部设备路由
	if err != nil {
		return fmt.Errorf("lookup routes uid=%d: %w", uid, err)
	}
	for _, rt := range routes { // routes 为空即全部离线，静默交给离线收件箱
		if err := d.push(ctx, rt, msg); err != nil {
			slog.Warn("deliver failed", "uid", uid, "conn_id", rt.ConnID, "err", err)
			if uerr := d.routes.Unbind(ctx, rt); uerr != nil { // 摘掉死连接的路由
				slog.Warn("unbind stale route", "conn_id", rt.ConnID, "err", uerr)
			}
		}
	}
	return nil // 单台设备失败不影响其他设备，重连后由 PullOffline 补齐
}

// push 单次投递：本机连接直接写发送队列，远端网关走 GatewayService.Deliver。
func (d *Deliverer) push(ctx context.Context, rt store.Route, msg *model.Message) error {
	if rt.GatewayID == d.selfID {
		return d.hub.Push(rt.UID, rt.ConnID, msg.AsPushFrame()) // 本机直投，不经过网络
	}
	cli, err := d.pool.Get(rt.GRPCAddr) // 按网关地址缓存的 gRPC 连接
	if err != nil {
		return fmt.Errorf("dial gateway %s: %w", rt.GRPCAddr, err)
	}
	ctx, cancel := context.WithTimeout(ctx, 300*time.Millisecond)
	defer cancel()
	_, err = cli.Deliver(ctx, &pb.DeliverReq{Uid: rt.UID, ConnId: rt.ConnID, MsgId: msg.MsgID,
		ConvId: msg.ConvID, Seq: msg.Seq, FromUid: msg.FromUID, Payload: msg.Payload})
	if err != nil {
		return fmt.Errorf("deliver to %s: %w", rt.GRPCAddr, err)
	}
	return nil
}

网关侧实现同一个 proto 里的 GatewayService.Deliver：按 (uid, connID) 在 Hub 里找到 Client，用第 03 章的信封封装一条 CmdPush 帧塞进 send 通道；找不到就返回 ok=false，逻辑层据此摘路由。这样「本机直投」与「跨网关投递」最后都落在同一条 send 队列上。

ACK 与回执：网关收到客户端的 ack 帧后，从待确认队列移除该消息并推进设备位点（第 09、10 章），同时给发送方推一条 ack 帧表示「对端已收到」；重复 ACK 在队列里找不到成员，直接返回成功，不会重复推回执。

验证方式

# 终端 1、2 用第 03 章的协议各连一个用户（第 02 章的 /ws 路由）
wscat -c "ws://127.0.0.1:8080/ws?token=<A的token>&device_id=a1"
wscat -c "ws://127.0.0.1:8080/ws?token=<B的token>&device_id=b1"
# A 发送 send_msg 帧（B 回 ACK 后 A 才会收到 receipt 回执）
{"cmd":"send_msg","ts":1735689600000,"body":{"client_msg_id":"c-1001","to":1002,"text":"hello"}}
# A 期望：{"cmd":"send_msg","msg_id":"70001","conv_id":"88001","seq":1,"code":0}
# B 期望：{"cmd":"push","msg_id":"70001","conv_id":"88001","seq":1,"from":1001}
# A 期望：{"cmd":"ack","msg_id":"70001","code":0}  回执
# conv_id 88001 落在 msg_01 分片（conv_id % 16，第 11 章），且只应有一条
mysql -uroot -p im -e "SELECT msg_id,conv_id,seq,from_uid FROM msg_01 WHERE conv_id=88001 ORDER BY seq DESC LIMIT 3;"

常见坑

坑	现象	正确做法
先投递再落库	进程崩溃就真丢消息，且无法补	先落库拿到 `msg_id` 再投递，失败转离线
收到上行就立刻回 ACK 帧	客户端以为成功，库里其实没有	回包必须在落库成功之后
单聊用 uid 拼 `conv_id`	后续会话级功能全部要写两套分支	单聊也走会话表，`conv_id` 由 `conv` 分配（第 11 章）
投递失败仍保留路由	后续消息持续投向死连接	失败即 `Unbind` 并转离线兜底

小结：单聊链路按「上行 → 落库 → 路由 → 投递 → ACK → 回执」推进，每跳的成功语义各自独立；上行靠 client_msg_id 唯一索引保证重发安全，投递靠「多端扇出 + 失败摘路由 + 离线兜底」保证最终可达，回执只改发送方的 UI 状态而不影响消息本身。下一章把 to_uid 换成 group_id，看群聊怎么扩散。

笔记加载中…

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

返回知识库