Go 企业 IM 架构消息可靠投递和离线推送的工程方案一、一条消息发出去对方说没收到企业 IM 最不能出错的就是消息可靠性。但现实中消息丢失的场景远比想象的多用户手机切后台时 WebSocket 断连消息还在服务端内存里服务重启时内存中的待推送消息全部丢失跨机房网络抖动导致消息路由到错误的节点。这些问题在外表看起来很偶发但在 10 万日活的场景下每天可能出现几百次。企业 IM 的消息可靠性不是每条消息都要保证 100% 送达那不现实而是消息不丢不重丢了也要能被发现和恢复。这是 IM 系统最难也最重要的设计目标。二、消息可靠投递的架构设计核心设计包括消息持久化优先、ACK 确认机制、离线消息补偿关键设计消息在推送之前必须已落盘先写 MySQL再推送。如果推送失败用户不在线消息已经在数据库中用户上线时可以拉取。这样即使消息服务崩溃重启也不丢消息。三、Go 实现消息服务核心逻辑package im import ( context database/sql encoding/json fmt sync time github.com/go-redis/redis/v8 ) // MessageStatus 消息状态 type MessageStatus int const ( StatusSent MessageStatus iota // 已发送已落盘 StatusDelivering // 推送中 StatusDelivered // 已送达 StatusRead // 已读 StatusFailed // 推送失败 ) // Message 消息结构 type Message struct { MsgID string json:msg_id // 全局唯一ID SeqID int64 json:seq_id // 用户递增序号 FromUserID string json:from_user_id ToUserID string json:to_user_id GroupID string json:group_id,omitempty Content string json:content ContentType string json:content_type // text, image, file Timestamp time.Time json:timestamp Status MessageStatus json:status } // IMService IM 核心服务 type IMService struct { db *sql.DB redis *redis.Client // 用户在线网关映射: userID - gatewayAddr onlineMap sync.Map // 递增序列号生成器 seqGen *SequenceGenerator } // NewIMService 创建 IM 服务 func NewIMService(db *sql.DB, redis *redis.Client) *IMService { return IMService{ db: db, redis: redis, seqGen: NewSequenceGenerator(redis), } } // SendMessage 发送消息核心流程 func (s *IMService) SendMessage( ctx context.Context, msg *Message, ) error { if msg.FromUserID || (msg.ToUserID msg.GroupID ) { return fmt.Errorf(消息发送方或接收方为空) } // 第 1 步生成消息ID和序号 msg.MsgID generateMsgID() var err error msg.SeqID, err s.seqGen.Next( ctx, msg.ToUserID, ) if err ! nil { return fmt.Errorf(生成序号失败: %w, err) } msg.Timestamp time.Now() msg.Status StatusSent // 第 2 步持久化到 MySQL同步保证不丢 if err : s.persistMessage(ctx, msg); err ! nil { return fmt.Errorf(消息持久化失败: %w, err) } // 第 3 步异步推送不阻塞发送方 go s.deliverMessage(context.Background(), msg) return nil } // persistMessage 持久化消息到 MySQL func (s *IMService) persistMessage( ctx context.Context, msg *Message, ) error { query : INSERT INTO messages (msg_id, seq_id, from_user_id, to_user_id, group_id, content, content_type, timestamp, status) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) _, err : s.db.ExecContext(ctx, query, msg.MsgID, msg.SeqID, msg.FromUserID, msg.ToUserID, msg.GroupID, msg.Content, msg.ContentType, msg.Timestamp, msg.Status, ) if err ! nil { return fmt.Errorf(数据库写入失败: %w, err) } return nil } // deliverMessage 推送消息给接收方 func (s *IMService) deliverMessage( ctx context.Context, msg *Message, ) { // 更新状态为推送中 s.updateMessageStatus(ctx, msg.MsgID, StatusDelivering) // 检查接收方是否在线 gatewayAddr, online : s.onlineMap.Load(msg.ToUserID) if !online { // 用户离线写入离线消息队列 s.storeOfflineMessage(ctx, msg) s.pushOfflineNotification(ctx, msg.ToUserID) return } // 用户在线推送到对应网关 delivered : s.pushToGateway( ctx, gatewayAddr.(string), msg, ) if delivered { s.updateMessageStatus(ctx, msg.MsgID, StatusDelivered) } else { // 推送失败标记为失败并加入离线队列 s.updateMessageStatus(ctx, msg.MsgID, StatusFailed) s.storeOfflineMessage(ctx, msg) } } // pushToGateway 推送消息到接入网关 func (s *IMService) pushToGateway( ctx context.Context, gatewayAddr string, msg *Message, ) bool { // 实际项目中通过 RPC 或消息队列推送到接入层 data, _ : json.Marshal(msg) err : s.redis.Publish( ctx, fmt.Sprintf(gateway:%s:push, gatewayAddr), data, ).Err() return err nil } // storeOfflineMessage 存储离线消息 func (s *IMService) storeOfflineMessage( ctx context.Context, msg *Message, ) { key : fmt.Sprintf(offline_msgs:%s, msg.ToUserID) msgJSON, _ : json.Marshal(msg) // ZSet: score 为 seq_idmember 为消息 JSON s.redis.ZAdd(ctx, key, redis.Z{ Score: float64(msg.SeqID), Member: msgJSON, }) // 设置 7 天过期 s.redis.Expire(ctx, key, 7*24*time.Hour) } // PullOfflineMessages 拉取离线消息用户上线时调用 func (s *IMService) PullOfflineMessages( ctx context.Context, userID string, limit int64, ) ([]*Message, error) { key : fmt.Sprintf(offline_msgs:%s, userID) // 按 seq_id 升序拉取 members, err : s.redis.ZRange(ctx, key, 0, limit-1).Result() if err ! nil { return nil, fmt.Errorf(拉取离线消息失败: %w, err) } var messages []*Message for _, member : range members { var msg Message if json.Unmarshal([]byte(member), msg) nil { messages append(messages, msg) } } // 从队列中移除已拉取的消息 if len(messages) 0 { s.redis.ZRemRangeByRank(ctx, key, 0, int64(len(messages)-1)) } return messages, nil } // pushOfflineNotification 发送离线推送通知APNs/FCM func (s *IMService) pushOfflineNotification( ctx context.Context, userID string, ) { // 实际项目中调用推送服务APNs iOS / FCM Android fmt.Printf(发送离线通知给用户 %s\n, userID) } // updateMessageStatus 更新消息状态 func (s *IMService) updateMessageStatus( ctx context.Context, msgID string, status MessageStatus, ) { query : UPDATE messages SET status ? WHERE msg_id ? s.db.ExecContext(ctx, query, status, msgID) } // SequenceGenerator 用户消息序号生成器 type SequenceGenerator struct { redis *redis.Client } func NewSequenceGenerator(redis *redis.Client) *SequenceGenerator { return SequenceGenerator{redis: redis} } func (sg *SequenceGenerator) Next( ctx context.Context, userID string, ) (int64, error) { key : fmt.Sprintf(seq:%s, userID) return sg.redis.Incr(ctx, key).Result() } func generateMsgID() string { // 雪花算法或 UUID return fmt.Sprintf(msg_%d, time.Now().UnixNano()) }四、边界分析与 Trade-offs同步写 DB 的性能代价每条消息同步写 MySQL 会增加 5-10ms 的延迟。在高并发场景如群发 5000 人的群下如果每条消息都同步写吞吐量可能不够。解决方案单聊消息同步写数据量小可靠性要求高群聊消息异步批量写数据量大可以容忍 100ms 内的延迟。离线消息的存储策略Redis ZSet 适合做离线消息的临时队列但内存有限。离线时间超过 7 天仍未被拉取的消息应该降级到 MySQL 或对象存储中用户点查看更早消息时再拉取。消息去重 vs 投递保证移动网络下客户端可能重复发送同一消息TCP 重传。服务端通过 msg_id 做幂等处理相同 msg_id 只处理一次配合客户端生成唯一 msg_id。但不要用消息内容哈希做去重——用户可能真的想发两条相同的消息。已读回执的批量更新每条消息都发一次已读回执写 DB 的压力太大。客户端应批量提交已读回执如每 500ms 或攒够 20 条服务端用UPDATE ... WHERE msg_id IN (...)批量更新。五、总结企业 IM 的消息可靠性核心原则先落盘后推送。数据库是消息的Source of Truth内存和 Redis 是加速层。离线消息用 Redis ZSet 做临时队列按 seq_id 排序、支持批量拉取、自动过期超过 7 天不拉取的消息降级到 MySQL。推送失败不能静默——要写入失败日志并触发重试还要通知发送方消息未送达到目标不是发送失败。消息的 seq_id 服务端生成Redis INCR避免客户端排序导致的乱序问题。
)
