Files
bl/common/socket/ServerEvent.go
昔念 b25f033939
All checks were successful
ci/woodpecker/push/my-first-workflow Pipeline was successful
```
refactor(socket): 移除重复保存锁并优化panic错误处理

移除了ServerEvent.go中的sync.Once保存锁,因为可能存在重复保存问题,
同时将panic错误处理从fmt.Println改为使用cool.Logger.Error进行统一日志记录

feat(player): 优化踢人功能并添加超时机制

移除kick.go中的复杂异步超时逻辑
2026-02-02 23:11:14 +08:00

219 lines
5.1 KiB
Go
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

package socket
import (
"context"
"fmt"
"log"
"os"
"sync/atomic"
"time"
"blazing/common/socket/codec"
"blazing/cool"
"blazing/logic/service/player"
"github.com/gogf/gf/v2/frame/g"
"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 { // 恢复 panicerr 为 panic 错误值
// 1. 打印错误信息
fmt.Println(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.Player.Save() //保存玩家数据
}
//}
//关闭连接
return
}
func (s *Server) OnTick() (delay time.Duration, action gnet.Action) {
g.Log().Async().Info(context.Background(), gtime.Now().ISO8601(), cool.Config.ServerInfo.OnlineID, "链接数", 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
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 { // 恢复 panicerr 为 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
return gnet.Close
}
}
return action
}
// CROSS_DOMAIN 定义跨域策略文件内容
const CROSS_DOMAIN = "<?xml version=\"1.0\"?><!DOCTYPE cross-domain-policy><cross-domain-policy><allow-access-from domain=\"*\" to-ports=\"*\" /></cross-domain-policy>\x00"
// TEXT 定义跨域请求的文本格式
const TEXT = "<policy-file-request/>\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
}