第20章 · 即时通讯 IM 服务（正式版）

GET /ws HTTP/1.1
Upgrade: websocket
Connection: Upgrade
Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==    # 随机串，用于校验握手合法性
Sec-WebSocket-Protocol: chat                    # 子协议（可选）
Sec-WebSocket-Version: 13                       # 协议版本
Origin: https://example.com                     # 来源（跨域控制）

HTTP/1.1 101 Switching Protocols
Upgrade: websocket
Connection: Upgrade
Sec-WebSocket-Accept: s3pPLMBiTxaQ9kYGzzhZRbK+xOo=   # 由 Key 计算得出

go get github.com/gorilla/websocket

// ws_demo/main.go（第一块：升级器）
package main

import (
    "fmt"
    "net/http"
    "time"

    "github.com/gorilla/websocket"
)

// Upgrader 把 HTTP 连接升级为 WebSocket
var upgrader = websocket.Upgrader{
    ReadBufferSize:  1024, // 读缓冲区大小
    WriteBufferSize: 1024, // 写缓冲区大小
    // CheckOrigin 校验请求来源。演示放行所有来源，生产必须校验白名单！
    CheckOrigin: func(r *http.Request) bool { return true },
}

// ws_demo/main.go（第二块：处理器与入口）
func handleWS(w http.ResponseWriter, r *http.Request) {
    conn, err := upgrader.Upgrade(w, r, nil) // ① HTTP → WebSocket
    if err != nil {
        return
    }
    defer conn.Close()

    go func() { // ② 独立 goroutine：每秒主动推一条（体会"服务端主动说"）
        for {
            if conn.WriteMessage(websocket.TextMessage,
                []byte(fmt.Sprintf(`{"ts":%d}`, time.Now().Unix()))) != nil {
                return
            }
            time.Sleep(time.Second)
        }
    }()

    for { // ③ 读循环：收客户端发来的消息
        _, msg, err := conn.ReadMessage()
        if err != nil {
            return
        }
        fmt.Printf("received: %s\n", msg)
    }
}

func main() {
    http.HandleFunc("/ws", handleWS)
    http.ListenAndServe(":8600", nil)
}

go run .
# 另开一个终端，安装 wscat（WebSocket 版的 curl）
npm install -g wscat
wscat -c ws://localhost:8600/ws
# 预期：每秒打印一条 {"ts":1712345678}——这就是服务端主动推送；
# 你手动敲 "hello" 回车，服务端终端打印 received: hello

CheckOrigin: func(r *http.Request) bool {
    return r.Header.Get("Origin") == "https://your-site.com"
}

var upgrader = websocket.Upgrader{
    WriteBufferPool: &sync.Pool{
        New: func() any { return make([]byte, 1024) },
    },
}

// ws_demo/hub.go（第一块：结构体与上下线）
package ws

import (
    "sync"

    "github.com/gorilla/websocket"
)

// Hub 维护所有在线连接，实现广播
type Hub struct {
    mu    sync.RWMutex          // Go 内置 map 非线程安全，读写都要加锁
    conns map[int64]*websocket.Conn // uid -> conn
}

func NewHub() *Hub {
    return &Hub{conns: make(map[int64]*websocket.Conn)}
}

// Register 用户上线时注册连接
func (h *Hub) Register(uid int64, conn *websocket.Conn) {
    h.mu.Lock()
    defer h.mu.Unlock()
    h.conns[uid] = conn
}

// Unregister 用户下线时移除
func (h *Hub) Unregister(uid int64) {
    h.mu.Lock()
    defer h.mu.Unlock()
    delete(h.conns, uid)
}

// ws_demo/hub.go（第二块：广播）
// Broadcast 广播消息，skipUid 用于不转发给发送者自己
func (h *Hub) Broadcast(msg []byte, skipUid int64) {
    h.mu.RLock() // 只读遍历，用读锁——多个广播可并发
    defer h.mu.RUnlock()
    for uid, conn := range h.conns {
        if uid == skipUid {
            continue // 不转发给自己
        }
        // 注意：生产中写入要异步 + 超时控制，见下面的坑
        _ = conn.WriteMessage(websocket.TextMessage, msg)
    }
}

// im-demo/message.go
package im

// Message IM 消息格式（教学最简版）
type Message struct {
    Seq     int64  `json:"seq"`     // 前端生成的序列号，本次连接内唯一即可
    Type    string `json:"type"`    // 消息类型：text/image/system...
    Content string `json:"content"` // 具体内容，根据 Type 解析
    Cid     int64  `json:"cid"`     // channel id：聊天 ID（单聊/群聊/系统通知）
}

// im-demo/gateway.go（第一块：读循环与转发）
package im

import (
    "context"
    "encoding/json"

    "github.com/gorilla/websocket"
)

// Gateway WebSocket 网关
type Gateway struct {
    conn    *websocket.Conn
    uid     int64
    backend BackendService // 转发到后端服务
}

// ReceiveAndForward 接收客户端消息并转发到后端
func (g *Gateway) ReceiveAndForward() error {
    for {
        _, data, err := g.conn.ReadMessage() // ① 读
        if err != nil {
            return err
        }
        var msg Message
        if err := json.Unmarshal(data, &msg); err != nil {
            g.sendError(0, "invalid message format") // ② 格式错，通知前端
            continue
        }
        if err := g.backend.SendMessage(context.Background(), g.uid, msg); err != nil {
            g.sendError(msg.Seq, "send failed") // ③ 转发失败，通知前端重试
            continue
        }
        g.sendAck(msg.Seq) // ④ 成功，回 ACK（3.8 节展开）
    }
}

// im-demo/gateway.go（第二块：ACK 与错误帧）
func (g *Gateway) sendAck(seq int64) {
    ack, _ := json.Marshal(map[string]any{"type": "ack", "seq": seq})
    _ = g.conn.WriteMessage(websocket.TextMessage, ack)
}

func (g *Gateway) sendError(seq int64, reason string) {
    e, _ := json.Marshal(map[string]any{"type": "error", "seq": seq, "reason": reason})
    _ = g.conn.WriteMessage(websocket.TextMessage, e)
}

// im-demo/backend.go
package im

import "context"

// BackendService 后端服务接口
type BackendService interface {
    SendMessage(ctx context.Context, sender int64, msg Message) error
}

// KafkaBackend 基于 Kafka 的后端实现
type KafkaBackend struct {
    groupSvc GroupService  // 查询群成员
    producer KafkaProducer // Kafka 生产者
}

func (b *KafkaBackend) SendMessage(ctx context.Context, sender int64, msg Message) error {
    // ① 找到当前聊天的所有成员
    members, err := b.groupSvc.GetMembers(ctx, msg.Cid)
    if err != nil {
        return err
    }
    // ② 投递到 Kafka，每个网关节点都会消费到
    //    生产中这里还会做：存储消息记录、内容审核、同步搜索等
    return b.producer.Send("msg", encodeMsg(sender, msg, members))
}

// im-demo/consume.go
package im

import "context"

// ConsumeLoop 每个网关节点都作为独立消费组，消费全部消息
func (g *Gateway) ConsumeLoop(ctx context.Context) error {
    // 关键：以「节点 ID」为消费组名 → 每个网关都是独立消费组，各拿全量消息
    consumer := g.kafka.Consumer("msg-group-" + g.nodeID)
    for {
        select {
        case <-ctx.Done():
            return ctx.Err()
        case msg := <-consumer.Messages():
            var event MsgEvent
            json.Unmarshal(msg.Value, &event)
            // 只把连在自己节点上的目标用户真正下发
            for _, uid := range event.TargetUids {
                if conn, ok := g.hub.Get(uid); ok {
                    _ = conn.WriteMessage(websocket.TextMessage, event.Payload)
                }
            }
        }
    }
}

// 心跳：读超时 + Pong 续期 + 定时 Ping（三件套）
// 步骤 1：设置读超时。超过 readDeadline 还没收到任何东西（含 Pong），
// ReadMessage 会返回超时错误 → 走断开清理逻辑
conn.SetReadDeadline(time.Now().Add(90 * time.Second))
conn.SetPongHandler(func(string) error {
    // 收到 Pong，读超时"续命"
    conn.SetReadDeadline(time.Now().Add(90 * time.Second))
    return nil
})

// 步骤 2：另起 goroutine 定时发 Ping（30s 一次）
go func() {
    ticker := time.NewTicker(30 * time.Second)
    defer ticker.Stop()
    for range ticker.C {
        if err := conn.WriteMessage(websocket.PingMessage, nil); err != nil {
            return // 写失败，说明连接已死
        }
    }
}()

# 克隆部署仓库
git clone https://github.com/openimsdk/openim-docker openim-docker
cd openim-docker
make init  # 初始化部署参数

# 配置 OPENIM_IP（用环境变量）
export OPENIM_IP=<你的服务器 IP>

# 启动（建议先清理 Docker 中同名容器，避免端口冲突）
docker-compose up -d

// internal/openim/syncer.go（第一块：结构体与请求构造）
package openim

import (
    "context"
    "net/http"

    "github.com/gotomicro/ekit/httpx"
)

// UserSyncer 把业务系统用户同步到 OpenIM
type UserSyncer struct {
    secret    string // 部署 OpenIM 时配置在 config/config.yaml 的 secret
    apiBase   string // 例如 http://OPENIM_IP:10002
    transport http.RoundTripper
}

// SyncUser 注册用户到 OpenIM
func (s *UserSyncer) SyncUser(ctx context.Context, uid int64, nickname, phone string) error {
    reqBody := map[string]any{
        "secret": s.secret,
        "users": []map[string]any{{
            "userID":   uid,
            "nickname": nickname,
            "phone":    phone,
        }},
    }
    req, err := httpx.NewRequest(ctx, http.MethodPost, s.apiBase+"/user/user_register", reqBody)
    if err != nil {
        return err
    }
    // operationID 标识本次请求，方便链路追踪——推荐用 OTel 的 trace ID
    req.Header.Set("operationID", traceIDFromContext(ctx))
    return s.do(req)
}

// internal/openim/syncer.go（第二块：发送与校验）
func (s *UserSyncer) do(req *http.Request) error {
    var resp struct {
        ErrCode int    `json:"errCode"`
        ErrMsg  string `json:"errMsg"`
    }
    if err := httpx.DoRequest(s.transport, req, &resp); err != nil {
        return err
    }
    if resp.ErrCode != 0 { // OpenIM 的约定：ErrCode 不为 0 即失败
        return fmt.Errorf("openim register failed: %s", resp.ErrMsg)
    }
    return nil
}

// im-service/internal/domain/message.go
package domain

import "encoding/json"

// Message 消息体：幂等 ID + 会话类型 + 状态位的生产规格
type Message struct {
    MsgID      string `json:"msg_id"`      // 消息唯一 ID（客户端生成的 UUID），服务端按它幂等去重
    MsgType    string `json:"msg_type"`    // text / image / file / system
    ConvType   string `json:"conv_type"`   // single 私聊 / group 群聊
    SenderID   int64  `json:"sender_id"`
    ReceiverID int64  `json:"receiver_id"` // 私聊=对方用户 ID；群聊=群 ID
    Content    string `json:"content"`     // 按 MsgType 解释：文本/图片URL/文件URL…
    Timestamp  int64  `json:"timestamp"`   // Unix 秒
    Status     string `json:"status"`      // sending / sent / read / failed
}

// im-service/internal/domain/frame.go
package domain

// Frame.Type 常量：信令只做传输控制，不落库；message_read / sync_request 等留作练习
const (
    FrameMessage    = "message"     // 业务消息
    FrameConnectAck = "connect_ack" // 连接建立确认
    FrameHeartbeat  = "heartbeat"   // 应用层心跳（协议层 Ping/Pong 之外的兜底）
    FrameMessageAck = "message_ack" // 服务端已接收 / 已送达
    FrameError      = "error"       // 服务端错误提示
)

// Frame 传输帧：所有 WebSocket 报文的统一外壳
type Frame struct {
    Type    string          `json:"type"`              // 取上面的常量
    Payload json.RawMessage `json:"payload,omitempty"` // 按 Type 再解析
}

// ForwardEvent im-forward topic 的载荷：跨节点转发的定向包裹
type ForwardEvent struct {
    GatewayAddr string  `json:"gateway_addr"` // 接收者所在网关地址（查 im:route 得到）
    TargetUID   int64   `json:"target_uid"`   // 实际推送对象（群聊时 != Message.ReceiverID）
    Message     Message `json:"message"`
}

// im-service/internal/gateway/conn_manager.go（第一块：常量与 Conn）
package gateway

import (
    "errors"
    "sync"
    "time"

    "github.com/gorilla/websocket"
)

// 心跳参数：30s 一次 Ping，90s 收不到报文判死连接（3.9 节的工程约定）
const (
    pingInterval = 30 * time.Second
    readTimeout  = 90 * time.Second
    writeTimeout = 10 * time.Second // 单条写超时：慢消费者保护
    sendQueueLen = 256
)

var ErrSlowConsumer = errors.New("send queue full: slow consumer")

// Conn 单条连接：所有写（业务消息 + Ping）都投给 send 队列，
// 由唯一的 writeLoop 串行写出——gorilla/websocket 不允许并发写同一连接
type Conn struct {
    ws     *websocket.Conn
    send   chan []byte
    once   sync.Once
    closed chan struct{}
}

func newConn(ws *websocket.Conn) *Conn {
    c := &Conn{ws: ws, send: make(chan []byte, sendQueueLen), closed: make(chan struct{})}
    go c.writeLoop() // 启动唯一的写 goroutine
    return c
}

// im-service/internal/gateway/conn_manager.go（第二块：writeLoop）
// writeLoop 唯一的写 goroutine：业务写和 Ping 在这里汇合成串行流
func (c *Conn) writeLoop() {
    ticker := time.NewTicker(pingInterval)
    defer ticker.Stop()
    write := func(mt int, data []byte) bool { // 统一设写超时；写失败=连接已死
        _ = c.ws.SetWriteDeadline(time.Now().Add(writeTimeout))
        return c.ws.WriteMessage(mt, data) == nil
    }
    for {
        select {
        case msg := <-c.send:
            if !write(websocket.TextMessage, msg) {
                c.Close()
                return
            }
        case <-ticker.C:
            if !write(websocket.PingMessage, nil) {
                c.Close()
                return
            }
        case <-c.closed:
            _ = c.ws.Close()
            return
        }
    }
}

// TrySend 非阻塞投递：队列满=慢消费者，交给上层断开，绝不阻塞调用方
func (c *Conn) TrySend(msg []byte) error {
    select {
    case c.send <- msg:
        return nil
    default:
        return ErrSlowConsumer // 3.5 节「慢消费者拖垮 Hub」的根治方案
    }
}

// Close 幂等关闭（sync.Once 保证多个 goroutine 同时调也只关一次）
func (c *Conn) Close() { c.once.Do(func() { close(c.closed); _ = c.ws.Close() }) }

// im-service/internal/gateway/conn_manager.go（第三块：ConnManager）
// ConnManager 连接管理器。BroadcastAll（全量广播）/ CloseAll（关全部连接
// 并返回 uid，停机清路由用）同理：遍历 uid 逐个 Send/Close——先复制 uid
// 再放锁，持锁回调会死锁，留给你写
type ConnManager struct {
    mu    sync.RWMutex
    conns map[int64]map[*Conn]struct{} // uid -> 该用户的全部连接（多端）
}

func NewConnManager() *ConnManager {
    return &ConnManager{conns: make(map[int64]map[*Conn]struct{})}
}

func (m *ConnManager) Add(uid int64, c *Conn) { // 上线注册（第二次 Add 即多端登录）
    m.mu.Lock()
    defer m.mu.Unlock()
    if m.conns[uid] == nil {
        m.conns[uid] = make(map[*Conn]struct{})
    }
    m.conns[uid][c] = struct{}{}
}

// Remove 断开移除，最后一条没了删整个 entry；返回该用户是否还有连接
// （第 3 步据此决定要不要清 Redis 路由——多端在线时不能误清）
func (m *ConnManager) Remove(uid int64, c *Conn) bool {
    m.mu.Lock()
    defer m.mu.Unlock()
    delete(m.conns[uid], c)
    if len(m.conns[uid]) == 0 {
        delete(m.conns, uid)
        return false
    }
    return true
}

// Send 推给该用户全部连接（多端同步）；群聊批量下发就是外层循环调它
func (m *ConnManager) Send(uid int64, msg []byte) {
    m.mu.RLock()
    defer m.mu.RUnlock()
    for c := range m.conns[uid] {
        _ = c.TrySend(msg) // 非阻塞投递：慢消费者只牺牲自己
    }
}

// im-service/internal/gateway/ws_handler.go（第一块：依赖接口与结构体）
package gateway

import (
    "context"

    "github.com/redis/go-redis/v9"
    "github.com/yourname/content-community/im-service/internal/domain"
    "go.uber.org/zap"
)

// Producer 发 Kafka（第 4 步实现）；Offline 离线存取（第 5 步实现）
type Producer interface {
    ToMessage(ctx context.Context, msg domain.Message) error     // → im-message（key=sender_id）
    ToForward(ctx context.Context, ev domain.ForwardEvent) error // → im-forward（key=接收者，保序）
}

type Offline interface {
    Push(ctx context.Context, uid int64, msg domain.Message) error
    Drain(ctx context.Context, uid int64, send func(domain.Message) error) error
}

// Handler WebSocket 接入层
type Handler struct {
    manager  *ConnManager
    rdb      *redis.Client
    producer Producer
    offline  Offline
    nodeAddr string // 本节点地址：写进 Redis 路由，多实例部署时每实例不同
    jwtKey   []byte // JWT 密钥，与 user-service 共享（同第02章）
    logger   *zap.Logger
}

// upgrader 参数说明见正文 3.4 节；生产环境 CheckOrigin 务必校验白名单
var upgrader = websocket.Upgrader{CheckOrigin: func(r *http.Request) bool { return true }}

// im-service/internal/gateway/ws_handler.go（第二块：接入前半段）
// HandleWS 处理 ws://host:8600/ws?token=<jwt>
func (h *Handler) HandleWS(w http.ResponseWriter, r *http.Request) {
    // ① 升级前校验 JWT——升级成 WebSocket 后就没法回 HTTP 401 了（正文 3.3 节）
    uid, err := h.parseToken(r.URL.Query().Get("token"))
    if err != nil {
        http.Error(w, "invalid token", http.StatusUnauthorized)
        return
    }
    ws, err := upgrader.Upgrade(w, r, nil)
    if err != nil {
        return
    }
    conn := newConn(ws)
    h.manager.Add(uid, conn)
    routeKey := "im:route:" + strconv.FormatInt(uid, 10)
    defer func() { // 断开清理：Remove 返回 false = 该用户最后一条连接，才清路由
        if !h.manager.Remove(uid, conn) {
            _ = h.rdb.Del(context.Background(), routeKey).Err()
        }
    }()
    // ② 注册在线路由 im:route:<uid> = 本节点地址，TTL 60s（靠心跳续期）
    _ = h.rdb.Set(context.Background(), routeKey, h.nodeAddr, 60*time.Second).Err()
    // ③ 回 connect_ack；上线补拉离线消息在第 5 步接上
    ack, _ := domain.NewFrame(domain.FrameConnectAck, map[string]any{"uid": uid})
    _ = conn.TrySend(ack)
    h.serveLoop(uid, conn, ws, routeKey) // ④ 心跳 + 读循环，见第三块
}

// im-service/internal/gateway/ws_handler.go（第三块：心跳与读循环）
func (h *Handler) serveLoop(uid int64, conn *Conn, ws *websocket.Conn, routeKey string) {
    // 心跳落地（原理见正文 3.9 节）：Ping 由第 2 步的 writeLoop 发，
    // 这里只管读超时和续期
    ws.SetReadDeadline(time.Now().Add(readTimeout))
    ws.SetPongHandler(func(string) error {
        ws.SetReadDeadline(time.Now().Add(readTimeout)) // 收到 Pong，读超时续命
        _ = h.rdb.Set(context.Background(), routeKey, h.nodeAddr, 60*time.Second).Err() // 顺手续路由 TTL
        return nil
    })
    for { // 读循环：心跳直接回帧；业务消息进 im-message
        _, data, err := ws.ReadMessage()
        if err != nil {
            return // 读失败=连接关闭/超时，走 defer 清理
        }
        frame, err := domain.ParseFrame(data)
        if err != nil {
            continue // 解不开的帧直接丢弃
        }
        switch frame.Type {
        case domain.FrameHeartbeat:
            pong, _ := domain.NewFrame(domain.FrameHeartbeat, map[string]any{"ts": time.Now().Unix()})
            _ = conn.TrySend(pong)
        case domain.FrameMessage:
            var msg domain.Message
            if json.Unmarshal(frame.Payload, &msg) != nil {
                continue
            }
            msg.SenderID = uid     // 发送者以 token 为准，防止伪造 sender_id
            msg.Timestamp = time.Now().Unix()
            if err := h.producer.ToMessage(context.Background(), msg); err != nil {
                continue // 发送失败：回 error 帧让客户端重试（打包同上，略）
            }
            msg.Status = "sent" // message_ack：服务端已接收，客户端停止转圈/重试
            ackMsg, _ := domain.NewFrame(domain.FrameMessageAck, msg)
            _ = conn.TrySend(ackMsg)
        }
    }
}

wscat -c "ws://localhost:8600/ws?token=<jwt>"  # token 从 user-service 登录接口拿
# 预期：连上立刻收到 {"type":"connect_ack",...}；发 {"type":"heartbeat"} 预期收到带 ts 的回帧
redis-cli TTL im:route:1  # 在 1~60 之间跳动，每隔 30s 被 Pong 续回 60——不会归零

// im-service/internal/repository/producer.go
package repository

import (
    "encoding/json"
    "strconv"

    "github.com/IBM/sarama"
    "github.com/yourname/content-community/im-service/internal/domain"
)

// 双 topic 常量
const (
    TopicIMMessage = "im-message" // 网关 → 业务处理
    TopicIMForward = "im-forward" // 业务处理 → 目标网关（key=接收者，同用户保序）
)

// KafkaProducer 实现 gateway.Producer
type KafkaProducer struct{ p sarama.SyncProducer }

func (k *KafkaProducer) send(topic, key string, v any) error {
    data, err := json.Marshal(v)
    if err != nil {
        return err
    }
    // key 参与分区选择：同 key 进同分区、分区内有序——
    // 「按 receiver_id 分区保证顺序」就是这一行（正文 3.8 节的会话内有序）
    _, _, err = k.p.SendMessage(&sarama.ProducerMessage{
        Topic: topic, Key: sarama.StringEncoder(key), Value: sarama.ByteEncoder(data),
    })
    return err
}

func (k *KafkaProducer) ToMessage(ctx context.Context, msg domain.Message) error {
    return k.send(TopicIMMessage, strconv.FormatInt(msg.SenderID, 10), msg)
}

func (k *KafkaProducer) ToForward(ctx context.Context, ev domain.ForwardEvent) error {
    return k.send(TopicIMForward, strconv.FormatInt(ev.TargetUID, 10), ev)
}

// im-service/internal/service/message_service.go（第一块：结构与消费入口）
package service

import (
    "context"
    "encoding/json"

    "github.com/redis/go-redis/v9"
    "github.com/yourname/content-community/im-service/internal/domain"
    "go.uber.org/zap"
)

// Forwarder service 只需要转发能力（与 gateway.Producer 是同一个实现的两个面）
type Forwarder interface {
    ToForward(ctx context.Context, ev domain.ForwardEvent) error
}

// 消息存储与群成员查询（表结构同前几章风格，CRUD 自己写，第 6 步先用内存桩）
type MessageRepo interface{ Save(ctx context.Context, msg domain.Message) error }
type GroupRepo interface{ Members(ctx context.Context, groupID int64) ([]int64, error) }

// MessageService 消费 im-message：先落库，再决定「在线转发 / 离线暂存」。
// NewMessageService 就是逐依赖赋值，自己写
type MessageService struct {
    repo    MessageRepo
    groups  GroupRepo
    rdb     *redis.Client
    forward Forwarder
    offline Offline
    logger  *zap.Logger
}

// Consume 一条 im-message 的处理入口。返回 error → 不提交 offset →
// Kafka 重投（至少一次）；重复消息靠 msg_id 幂等
func (s *MessageService) Consume(data []byte) error {
    var msg domain.Message
    if err := json.Unmarshal(data, &msg); err != nil {
        s.logger.Error("poison message dropped", zap.Error(err)) // 毒消息重试也没用，记日志丢弃
        return nil
    }
    ctx := context.Background()
    // ① 先落库再转发：持久化成功才算"真的发出去了"（正文 3.8 节的先落库再 ACK）
    if err := s.repo.Save(ctx, msg); err != nil {
        return err
    }
    // ② 群聊从简：查成员逐人投递（发送者不回推，他已有 message_ack）
    if msg.ConvType == "group" {
        members, _ := s.groups.Members(ctx, msg.ReceiverID) // 生产要处理 err 与 >1000 人的异步批量
        for _, uid := range members {
            if uid != msg.SenderID {
                _ = s.deliver(ctx, msg, uid)
            }
        }
        return nil
    }
    return s.deliver(ctx, msg, msg.ReceiverID) // ③ 私聊：直接投给接收者
}

// im-service/internal/service/message_service.go（第二块：投递决策）
// deliver 投递决策：查在线路由——在线发 im-forward，离线存 Redis List
func (s *MessageService) deliver(ctx context.Context, msg domain.Message, uid int64) error {
    route, err := s.rdb.Get(ctx, "im:route:"+strconv.FormatInt(uid, 10)).Result()
    if err != nil && err != redis.Nil {
        return err // Redis 故障：交给 Kafka 重试，宁慢勿丢
    }
    if route != "" { // 在线：定向转发给持有连接的网关
        return s.forward.ToForward(ctx, domain.ForwardEvent{
            GatewayAddr: route, TargetUID: uid, Message: msg,
        })
    }
    return s.offline.Push(ctx, uid, msg) // 离线：暂存（第 5 步实现）
}

// im-service/internal/gateway/forward_consumer.go
package gateway

import (
    "encoding/json"

    "github.com/yourname/content-community/im-service/internal/domain"
)

// ConsumeForward 消费 im-forward 并下发给本节点的目标用户。
// 与正文 3.7 节同理：每个网关节点用「独立消费组」拿全量，
// 再按 gateway_addr 过滤——若共用一个消费组，消息会被随机某个
// 网关抢走，目标用户反而收不到（见下方坑）
func (h *Handler) ConsumeForward(data []byte) error {
    var ev domain.ForwardEvent
    if json.Unmarshal(data, &ev) != nil {
        return nil // 毒消息丢弃
    }
    if ev.GatewayAddr != h.nodeAddr {
        return nil // 不是投给本节点的，忽略
    }
    raw, err := domain.NewFrame(domain.FrameMessage, ev.Message)
    if err != nil {
        return err
    }
    h.manager.Send(ev.TargetUID, raw) // 推给该用户在本节点的全部连接（多端同步）
    return nil                        // 接收方的回执链路留作练习
}

{"type":"message","payload":{"msg_id":"u-001","msg_type":"text","conv_type":"single","receiver_id":2,"content":"你好"}}
# 预期：用户1 立刻收到 {"type":"message_ack",...,"status":"sent"}；
#       用户2 秒级收到 {"type":"message","payload":{...,"sender_id":1,"content":"你好",...}}
# 多端同步：用户2 再开第三个 wscat 连接，用户1 再发一条，用户2 的两个连接都收到

// im-service/internal/service/offline.go（第一块：Push）
package service

import (
    "context"
    "encoding/json"
    "fmt"
    "time"

    "github.com/redis/go-redis/v9"
    "github.com/yourname/content-community/im-service/internal/domain"
)

// OfflineRedis 离线消息实现：Redis List，key=im:offline:<uid>
type OfflineRedis struct{ rdb *redis.Client }

func NewOfflineRedis(rdb *redis.Client) *OfflineRedis { return &OfflineRedis{rdb: rdb} }

// Push 从右侧压入一条，并给整个 List 续 7 天过期（超期只剩 DB 里的）
func (o *OfflineRedis) Push(ctx context.Context, uid int64, msg domain.Message) error {
    data, err := json.Marshal(msg)
    if err != nil {
        return err
    }
    key := fmt.Sprintf("im:offline:%d", uid)
    pipe := o.rdb.Pipeline()
    pipe.RPush(ctx, key, data)
    pipe.Expire(ctx, key, 7*24*time.Hour)
    _, err = pipe.Exec(ctx)
    return err
}

// im-service/internal/service/offline.go（第二块：Drain）
// Drain 上线补拉：从左侧逐条弹出，推一条删一条；
// 中途断线，没弹出的还留在 List 里
func (o *OfflineRedis) Drain(ctx context.Context, uid int64, send func(domain.Message) error) error {
    key := fmt.Sprintf("im:offline:%d", uid)
    for {
        data, err := o.rdb.LPop(ctx, key).Bytes()
        if err == redis.Nil {
            return nil // 队列空，补拉完成
        }
        if err != nil {
            return err
        }
        var msg domain.Message
        if json.Unmarshal(data, &msg) != nil {
            continue // 脏数据跳过
        }
        if err := send(msg); err != nil { // 推送失败：塞回队首，下次上线再试
            o.rdb.LPush(ctx, key, data)
            return err
        }
    }
}

# 用户2 断开（wscat 里 Ctrl-C）；用户1 连发两条消息
redis-cli LLEN im:offline:2   # 2——离线暂存成功
# 用户2 重新连接：立刻按发送顺序收到两条离线消息
redis-cli LLEN im:offline:2   # 0——补拉后清空

# im-service/configs/config.yaml
im:
  ws_addr: ":8600"             # WebSocket 端口
  node_addr: "127.0.0.1:8600"  # 写进 Redis 路由的本节点地址；多实例部署时每实例不同
  jwt_secret: "dev-secret"     # 与 user-service 共享
kafka:
  brokers: [localhost:9092]
redis:
  addr: localhost:6379

// im-service/main.go（第一块：组装与消费组）
package main

import (
    "github.com/redis/go-redis/v9"
    "github.com/spf13/viper"
    "go.uber.org/zap"

    "github.com/yourname/content-community/im-service/internal/gateway"
    "github.com/yourname/content-community/im-service/internal/repository"
    "github.com/yourname/content-community/im-service/internal/service"
)

func main() {
    // 全局约定：viper 读 configs/ + zap 结构化日志；ETCD 注册同第12章，此处省略
    viper.AddConfigPath("configs")
    _ = viper.ReadInConfig()
    logger, _ := zap.NewProduction()
    rdb := redis.NewClient(&redis.Options{Addr: viper.GetString("redis.addr")})
    producer, err := repository.NewKafkaProducer(viper.GetStringSlice("kafka.brokers"))
    if err != nil {
        logger.Fatal("kafka producer failed", zap.Error(err))
    }
    offline := service.NewOfflineRedis(rdb)
    // 组装：同一个 producer/offline 实例，同时满足 gateway 与 service 两侧接口。
    // 两个 repo 参数教学版用内存桩（Save 存 slice、Members 查 map，各 3 行，自己写），
    // 换成前几章的 GORM 实现即是生产版
    manager := gateway.NewConnManager()
    handler := gateway.NewHandler(manager, rdb, producer, offline,
        viper.GetString("im.node_addr"), viper.GetString("im.jwt_secret"), logger)
    msgSvc := service.NewMessageService(&memMessageRepo{},
        &memGroupRepo{members: map[int64][]int64{1: {1, 2, 3}}}, rdb, producer, offline, logger)
    // 消费组一：im-message，全网一个组（任一 im-service 实例处理一次即可）
    go repository.MustConsume(viper.GetStringSlice("kafka.brokers"), "im-message-group",
        repository.TopicIMMessage, msgSvc.Consume, logger)
    // 消费组二：im-forward，每个网关独立消费组（组名带 node_addr，原因见第 4 步的坑）
    go repository.MustConsume(viper.GetStringSlice("kafka.brokers"),
        "im-forward-"+viper.GetString("im.node_addr"),
        repository.TopicIMForward, handler.ConsumeForward, logger)
    serve(manager, handler, logger, rdb) // HTTP 服务 + 优雅停机，见第二块
}

// im-service/main.go（第二块：HTTP 服务与优雅停机）
func serve(manager *gateway.ConnManager, handler *gateway.Handler,
    logger *zap.Logger, rdb *redis.Client) {
    mux := http.NewServeMux()
    mux.HandleFunc("/ws", handler.HandleWS)
    srv := &http.Server{Addr: viper.GetString("im.ws_addr"), Handler: mux}
    go func() { _ = srv.ListenAndServe() }()
    logger.Info("im-service started", zap.String("addr", srv.Addr))
    // 优雅停机：广播提示 → 等 5s 冲刷发送队列 → 断开并清路由
    quit := make(chan os.Signal, 1)
    signal.Notify(quit, syscall.SIGINT, syscall.SIGTERM)
    <-quit
    ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
    defer cancel()
    _ = srv.Shutdown(ctx) // 不再接受新连接
    closeFrame, _ := domain.NewFrame("system", domain.Message{
        MsgType: "system", Content: "server restarting, please reconnect",
        Timestamp: time.Now().Unix(),
    })
    manager.BroadcastAll(closeFrame) // 第 2 步练习：遍历 uid 逐个 Send
    time.Sleep(5 * time.Second)      // 给 writeLoop 时间把队列里的消息冲完
    for _, uid := range manager.CloseAll() { // 第 2 步练习：关全部连接并返回 uid
        rdb.Del(ctx, "im:route:"+strconv.FormatInt(uid, 10))
    }
    logger.Info("im-service stopped gracefully")
}

go run .
# 终端1 用户1、终端2 用户2 分别 wscat 连接
# ① 在线私聊：用户1 发 receiver_id=2 → 用户1 收 message_ack，用户2 秒级收到
# ② 离线暂存：用户2 断开，用户1 连发两条 → redis-cli LLEN im:offline:2 返回 2
# ③ 离线补拉：用户2 重连 → 按序收到两条，LLEN 归零
# ④ 群聊：conv_type=group、receiver_id=1 → 在线成员各收一份，发送者不重复收
# ⑤ 优雅停机：Ctrl-C → 客户端先收 system 提示再断开；redis-cli KEYS "im:route:*" 为空

content-community/                          # 本章改动一览（其余省略）
├── im-service/                             # 新增微服务（WebSocket :8600）
│   ├── configs/config.yaml                 # viper 配置：im / kafka / redis 段
│   ├── main.go                             # 组装 + 双消费组 + 优雅停机
│   └── internal/
│       ├── domain/
│       │   └── message.go                  # Message / Frame / ForwardEvent 协议（JSON 编解码）
│       ├── gateway/
│       │   ├── conn_manager.go             # Conn + ConnManager：多端连接管理、串行写、Ping（核心）
│       │   ├── ws_handler.go               # JWT 校验 + 路由注册 + 心跳 + 读循环（核心）
│       │   └── forward_consumer.go         # 消费 im-forward，按 gateway_addr 过滤下发
│       ├── service/
│       │   ├── message_service.go          # 消费 im-message：落库→查路由→转发/离线（核心）
│       │   └── offline.go                  # 离线消息 Redis List：Push + Drain
│       └── repository/
│           ├── producer.go                 # KafkaProducer：im-message / im-forward 双 topic
│           └── consumer.go                 # MustConsume：第07章消费组模板回调化
└── pkg/kafka/                              # 第07章已有，本章复用不改动