// Package netserver 服务器网络层:TCP 监听、连接会话、握手与心跳(多人同步.md、网络协议.md §4–§7)。 // // 线程模型(架构.md §4):每连接 1 读 + 1 写 goroutine;业务回调在主逻辑线程执行(通过 // 事件 channel 投递),tick 内不做 IO。 package netserver import ( "context" "errors" "fmt" "io" "net" "runtime/debug" "sync" "sync/atomic" "time" "mc/internal/logx" "mc/internal/netproto" ) // 连接超时与心跳参数(网络协议.md §7)。 const ( readTimeout = 30 * time.Second // 无响应判定断线 heartbeatIn = 5 * time.Second // 心跳间隔 maxFrameSize = netproto.MaxFrameSize ) // Handler 服务器业务回调(由主逻辑线程实现)。 type Handler interface { // OnConnect 连接建立(握手前)。 OnConnect(c *Conn) // OnHandshake 收到握手:返回登录结果(0 成功,其他为拒绝原因码)。 OnHandshake(c *Conn, hs netproto.Handshake) uint8 // OnMove 玩家移动(20Hz)。 OnMove(c *Conn, m netproto.PlayerMove) // OnDisconnect 连接断开。 OnDisconnect(c *Conn) } // Server TCP 服务器。 type Server struct { addr string log *logx.Logger handler Handler ln net.Listener mu sync.Mutex conns map[uint64]*Conn nextID atomic.Uint64 } // New 创建服务器(未监听)。 func New(addr string, log *logx.Logger, h Handler) *Server { return &Server{addr: addr, log: log, handler: h, conns: make(map[uint64]*Conn)} } // Listen 绑定端口并启动接受循环(ctx 取消即关闭)。 func (s *Server) Listen(ctx context.Context) error { ln, err := net.Listen("tcp", s.addr) if err != nil { return fmt.Errorf("netserver.Listen %s: %w", s.addr, err) } s.ln = ln s.log.Infof("网络监听于 %s", s.addr) go s.acceptLoop(ctx) return nil } // Addr 返回监听地址。 func (s *Server) Addr() string { return s.ln.Addr().String() } // acceptLoop 接受连接循环。 func (s *Server) acceptLoop(ctx context.Context) { go func() { <-ctx.Done() s.Close() }() for { nc, err := s.ln.Accept() if err != nil { return // 关闭监听即退出 } c := newConn(s.nextID.Add(1), nc, s) s.mu.Lock() s.conns[c.id] = c s.mu.Unlock() go c.readLoop() go c.writeLoop() } } // Close 关闭监听与全部连接(优雅退出)。 func (s *Server) Close() { if s.ln != nil { _ = s.ln.Close() } s.mu.Lock() conns := make([]*Conn, 0, len(s.conns)) for _, c := range s.conns { conns = append(conns, c) } s.mu.Unlock() for _, c := range conns { c.close() } } // Conn 一个客户端连接。 type Conn struct { id uint64 conn net.Conn srv *Server send chan []byte done chan struct{} closeOnce sync.Once } // newConn 创建连接。 func newConn(id uint64, nc net.Conn, s *Server) *Conn { return &Conn{id: id, conn: nc, srv: s, send: make(chan []byte, 64), done: make(chan struct{})} } // ID 连接 ID(服务器分配,多人同步.md §3)。 func (c *Conn) ID() uint64 { return c.id } // readLoop 读循环:解帧并分发(网络协议.md §1 帧格式)。 // 半包(ErrIncomplete)时保留缓冲区等待更多数据;协议错误直接断开。 func (c *Conn) readLoop() { defer c.srv.handler.OnDisconnect(c) buf := make([]byte, 0, 64*1024) tmp := make([]byte, 64*1024) for { _ = c.conn.SetReadDeadline(time.Now().Add(readTimeout)) n, err := c.conn.Read(tmp) if err != nil { return // 超时/断开:结束连接 } buf = append(buf, tmp[:n]...) c.srv.log.Debugf("连接 %d 读到 %d 字节(缓冲 %d)", c.id, n, len(buf)) if len(buf) > 2*maxFrameSize { c.srv.log.Warnf("连接 %d 缓冲区超限,断开", c.id) c.close() return } for len(buf) >= 4 { f, rest, err := netproto.DecodeFrame(buf) if err != nil { if errors.Is(err, netproto.ErrIncomplete) { break // 半包:保留 buf,等待更多数据 } c.srv.log.Warnf("连接 %d 帧解析失败: %v,断开", c.id, err) c.close() return } buf = rest c.handle(f) } } } // handle 消息分发(读 goroutine → 业务回调,业务侧自行保证线程安全)。 func (c *Conn) handle(f netproto.Frame) { switch f.MsgID { case netproto.MsgHandshake: hs, err := netproto.DecodeHandshake(f.Payload) if err != nil { c.close() return } code := c.srv.handler.OnHandshake(c, hs) c.Send(netproto.Frame{MsgID: netproto.MsgLoginResponse, Payload: []byte{code}}) if code != 0 { c.close() } case netproto.MsgPlayerMove: m, err := netproto.DecodePlayerMove(f.Payload) if err == nil { c.srv.handler.OnMove(c, m) } case netproto.MsgHeartbeat: c.Send(netproto.Frame{MsgID: netproto.MsgHeartbeat, Payload: f.Payload}) // 回显心跳 case netproto.MsgDisconnect: c.close() } } // writeLoop 写循环:channel 驱动,写超时断开(网络协议.md §7)。 func (c *Conn) writeLoop() { for { select { case <-c.done: return case data := <-c.send: _ = c.conn.SetWriteDeadline(time.Now().Add(readTimeout)) if _, err := c.conn.Write(data); err != nil { c.close() return } } } } // Send 发送一帧(非阻塞入队;连接关闭或队满时丢弃并告警,防慢连接阻塞服务器)。 func (c *Conn) Send(f netproto.Frame) { select { case <-c.done: return default: } select { case c.send <- netproto.EncodeFrame(f): case <-c.done: default: c.srv.log.Warnf("连接 %d 发送队列已满,丢弃帧 0x%02X", c.id, f.MsgID) } } // close 幂等关闭:关闭 done 信号与 socket;不关闭 send(避免与在途 Send 竞争)。 func (c *Conn) close() { c.closeOnce.Do(func() { c.srv.log.Debugf("连接 %d 关闭(来源:\n%s)", c.id, debug.Stack()) close(c.done) _ = c.conn.Close() c.srv.mu.Lock() delete(c.srv.conns, c.id) c.srv.mu.Unlock() }) } // Count 当前连接数(测试/统计)。 func (s *Server) Count() int { s.mu.Lock() defer s.mu.Unlock() return len(s.conns) } var _ = io.EOF // 保持 io 引用(后续扩展流式读取)