[network] Go 长连接服务的三种读写并发模型

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
type Conn struct {
    ws *websocket.Conn
    mu sync.Mutex // 每连接一把写锁
}

// 任意 goroutine 都能调用，靠锁保证同一连接不并发写
func (c *Conn) Write(data []byte) error {
    c.mu.Lock()
    defer c.mu.Unlock()
    c.ws.SetWriteDeadline(time.Now().Add(5 * time.Second)) // 写超时，别漏
    return c.ws.WriteMessage(websocket.TextMessage, data)
}

// 唯一的读 goroutine
func (c *Conn) readLoop() {
    for {
        _, msg, err := c.ws.ReadMessage()
        if err != nil {
            return
        }
        if string(msg) == "ping" {
            c.Write([]byte("pong")) // 读协程里也走加锁写
        }
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
type Conn struct {
    ws   *websocket.Conn
    send chan []byte // 带缓冲发送队列
}

// 非阻塞投递；满了 = 客户端太慢 -> 丢弃并关闭（背压）
func (c *Conn) Push(data []byte) {
    select {
    case c.send <- data:
    default:
        close(c.send) // 慢客户端，踢掉，不拖累别人
    }
}

// 读协程：只读，把要回的内容丢进队列，不直接写
func (c *Conn) readPump() {
    for {
        _, msg, err := c.ws.ReadMessage()
        if err != nil {
            return
        }
        if string(msg) == "ping" {
            c.Push([]byte("pong"))
        }
    }
}

// 写协程：本连接唯一 writer，无需锁；顺带统一发心跳
func (c *Conn) writePump() {
    tick := time.NewTicker(30 * time.Second)
    defer tick.Stop()
    for {
        select {
        case data, ok := <-c.send:
            if !ok {
                return // send 已关闭
            }
            c.ws.SetWriteDeadline(time.Now().Add(5 * time.Second))
            if c.ws.WriteMessage(websocket.TextMessage, data) != nil {
                return
            }
        case <-tick.C:
            c.ws.WriteMessage(websocket.PingMessage, nil)
        }
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
// 每连接的上下文，挂在 gnet.Conn 上
type ConnContext struct {
    ConnId     string
    ClientID   string
    Addr       string
    StartTime  time.Time
    IsConnOpen bool
}

type Server struct {
    gnet.BuiltinEventEngine
    eng gnet.Engine
}

// 新连接：建上下文 + 注册进连接管理器
func (s *Server) OnOpen(c gnet.Conn) (out []byte, action gnet.Action) {
    ctx := &ConnContext{
        ConnId:     genUUID(),
        Addr:       c.RemoteAddr().String(),
        StartTime:  time.Now(),
        IsConnOpen: true,
    }
    c.SetContext(ctx)                            // 状态挂到连接上
    clientMgr.Add(ctx.ConnId, c)                 // 注册，后面推送靠它找到 c
    return nil, gnet.None
}

// 可读：拆包 -> 处理 -> 回包，全程不阻塞
func (s *Server) OnTraffic(c gnet.Conn) gnet.Action {
    ctx, ok := c.Context().(*ConnContext)
    if !ok || !ctx.IsConnOpen {
        return gnet.Close
    }

    packet, err := readFullPacket(c)             // 自己粘包拆包，凑不齐一个整包就返回
    if err != nil {
        return gnet.Close
    }
    if packet == nil {
        return gnet.None                         // 半个包，等下次 OnTraffic
    }

    rsp := s.process(ctx, packet)                // 业务处理（注意：别在这里阻塞！）
    c.AsyncWrite(rsp, nil)                        // 异步回包，框架保证并发安全
    return gnet.None
}

// 断开：注销 + 清理，状态从 context 里捞
func (s *Server) OnClose(c gnet.Conn, err error) gnet.Action {
    ctx, ok := c.Context().(*ConnContext)
    if ok && ctx != nil {
        ctx.IsConnOpen = false
        clientMgr.Remove(ctx.ConnId)             // 先注销，别让推送再找到它
        clearCache(ctx.ConnId)                   // 清理会话密钥等缓存
        c.SetContext(nil)                        // 解引用，帮 GC
    }
    return gnet.Close
}