package socket import ( "context" "log" "os" "sync/atomic" "time" "blazing/common/socket/codec" "blazing/cool" "blazing/logic/service/player" "github.com/gogf/gf/v2/os/glog" "github.com/gogf/gf/v2/os/gtime" "github.com/panjf2000/gnet/v2" ) func (s *Server) Boot() error { // go s.bootws() err := gnet.Run(s, s.network+"://"+s.addr, gnet.WithMulticore(true), gnet.WithTicker(true), // gnet.WithReusePort(true), // gnet.WithReuseAddr(true), gnet.WithSocketRecvBuffer(s.bufferSize)) if err != nil { panic(err) } return nil } func (s *Server) Stop() error { _ = s.eng.Stop(context.Background()) s.workerPool.Release() return nil } func (s *Server) OnClose(c gnet.Conn, err error) (action gnet.Action) { defer func() { if err := recover(); err != nil { // 恢复 panic,err 为 panic 错误值 // 1. 打印错误信息 cool.Logger.Error(context.TODO(), "panic 错误:", err) } }() // 识别 RST 导致的连接中断(错误信息含 "connection reset") // if err != nil && (strings.Contains(err.Error(), "connection reset") || strings.Contains(err.Error(), "reset by peer")) { // remoteIP := c.RemoteAddr().(*net.TCPAddr).IP.String() // log.Printf("RST 攻击检测: 来源 %s, 累计攻击次数 %d", remoteIP) // // 防护逻辑:临时封禁异常 IP(可扩展为 IP 黑名单) // // go s.tempBlockIP(remoteIP, 5*time.Minute) // } //fmt.Println(err, c.RemoteAddr().String(), "断开连接") atomic.AddInt64(&s.connected, -1) //logging.Infof("conn[%v] disconnected", c.RemoteAddr().String()) v, _ := c.Context().(*player.ClientData) v.LF.Close() if v.Player != nil { v.SaveL.Do(func() { //使用保存锁,确保在踢人和掉线的时候只保存一次 //cool.Loger.Info(context.TODO(), "准备保存", v.Player.Info.UserID) v.Player.Save() //保存玩家数据 }) } //} //关闭连接 return } func (s *Server) OnTick() (delay time.Duration, action gnet.Action) { glog.Info(context.Background(), gtime.Now().ISO8601(), "链接数", atomic.LoadInt64(&s.connected)) if s.quit && atomic.LoadInt64(&s.connected) == 0 { //执行正常退出逻辑 os.Exit(0) } return 30 * time.Second, gnet.None } func (s *Server) OnBoot(eng gnet.Engine) gnet.Action { s.eng = eng // cool.Loger.Infof(context.Background(), " server is listening on %s\n", s.addr) return gnet.None } func (s *Server) OnOpen(conn gnet.Conn) (out []byte, action gnet.Action) { if s.network != "tcp" { return nil, gnet.Close } if conn.Context() == nil { conn.SetContext(player.NewClientData(conn)) //注入data } atomic.AddInt64(&s.connected, 1) return nil, gnet.None } func (s *Server) OnTraffic(c gnet.Conn) (action gnet.Action) { defer func() { if err := recover(); err != nil { // 恢复 panic,err 为 panic 错误值 // 1. 打印错误信息 cool.Logger.Error(context.TODO(), "panic 错误:", err) } }() ws := c.Context().(*player.ClientData).Wsmsg if ws.Tcp { //升级失败时候防止缓冲区溢出 return s.handleTcp(c) } tt, len1 := ws.ReadBufferBytes(c) if tt == gnet.Close { return gnet.Close } ok, action := ws.Upgrade(c) if action != gnet.None { //连接断开 return action } if !ok { //升级失败,说明是tcp连接 ws.Tcp = true return s.handleTcp(c) } // fmt.Println(ws.Buf.Bytes()) c.Discard(len1) messages, err := ws.Decode(c) if err != nil { return gnet.Close } if messages == nil { return } t := c.Context().(*player.ClientData) //client := conn.RemoteAddr().String() s.workerPool.Submit(func() { //TODO 这里可能存在顺序执行问题,待修复 for _, msg := range messages { t.LF.Producer().Write(msg.Payload) //t.OnEvent(msg.Payload) } }) return gnet.None } func (s *Server) handleTcp(conn gnet.Conn) (action gnet.Action) { conn.Context().(*player.ClientData).IsCrossDomain.Do(func() { //跨域检测 handle(conn) }) data, err := s.codec.Decode(conn) if err != nil { if err == codec.ErrIncompletePacket { return } return gnet.Close } s.workerPool.Submit(func() { //TODO 这里可能存在顺序执行问题,待修复 //conn.Context().(*player.ClientData).OnEvent(data) if t, ok := conn.Context().(*player.ClientData); ok { t.LF.Producer().Write(data) } }) if conn.InboundBuffered() > 0 { if err := conn.Wake(nil); err != nil { // wake up the connection manually to avoid missing the leftover data cool.Logger.Errorf(context.Background(), "failed to wake up the connection, %v", err) return gnet.Close } } return action } // CROSS_DOMAIN 定义跨域策略文件内容 const CROSS_DOMAIN = "\x00" // TEXT 定义跨域请求的文本格式 const TEXT = "\x00" func handle(c gnet.Conn) { // 读取数据并检查是否为跨域请求 data, err := c.Peek(len(TEXT)) if err != nil { log.Printf("Error reading cross-domain request: %v", err) return } if string(data) == TEXT { //判断是否是跨域请求 log.Printf("Received cross-domain request from %s", c.RemoteAddr()) // 处理跨域请求 c.Write([]byte(CROSS_DOMAIN)) c.Discard(len(TEXT)) return } //return }