package baseserver import ( "rocommon" _ "rocommon/rpc" "rocommon/service" "rocommon/socket" "rocommon/util" "roserver/baseserver/model" "roserver/serverproto" "runtime" "sort" "strconv" "time" ) var Queue rocommon.NetEventQueue type ServiceParam struct { ServiceName string ServiceType string //tcpAcceptor /tcpConnector ProcName string LisAddr string DiscoveryServiceName string //用于服务器发现 DiscoveryServiceZone int //用于服务器发现 } //loadConfig是否需要加载配置文件, update是否有主循环 func Init(serverName string, loadConfig func(), update rocommon.UpdateModule) { //设置携程使用的线程数量(不会强制性生效) //The GOMAXPROCS variable limits the number of operating system threads that can execute user-level Go code simultaneously. // There is no limit to the number of threads that can be blocked in system calls on behalf of Go code; those do not count // against the GOMAXPROCS limit. This package's GOMAXPROCS function queries and changes the limit. runtime.GOMAXPROCS(runtime.NumCPU() / 2) //dgnet -> init.go service.Init(serverName) util.InfoF("////////////////////////Service GameVersion=%v DebugMode=%v", int32(serverproto.GameVersion_GameVersion_Main), service.DebugMode) if loadConfig != nil { loadConfig() } //事件回调队列 Queue = service.NewEventQueue() //设置更新操作 if update != nil { Queue.AttachUpdateModule(update) } Queue.StartQueue() //链接服务器发现 etcd 在service.Init中已经实现 //log.Printf("server init ok...") util.InfoF("server init ok...") } func Wait() { //todo... //监听退出信号 service.WaitExitSignal() Queue.StopQueue() Queue.Wait() } func Exit(node rocommon.ServerNode) { //todo... 停止所有在监听的协程 util.InfoF("exit...") //停止etcd if node != nil { node.Stop() service.ETCDUnregister(node) } service.GetServiceDiscovery().Close() } var idxMap = map[int]*model.PerformKVTimeSt{} var performTime uint64 = 0 func KVTimeTest(nowTime uint64) { var printfList []model.PerformKVTimeSt if performTime <= 0 { performTime = nowTime } else if nowTime-performTime >= 1000 { performTime = nowTime for _, val := range idxMap { printfList = append(printfList, *val) } } if len(printfList) > 0 { sort.Slice(printfList, func(i, j int) bool { return printfList[i].AckMsgId < printfList[j].AckMsgId }) printfListStr := "" for idx := 0; idx < len(printfList); idx++ { //msgIdStr := strconv.Itoa(int(printfList[idx].AckMsgId)) tmpTime := float64(printfList[idx].TotalTime) / float64(printfList[idx].TotalNum) tmpPerNum := printfList[idx].NowTime - printfList[idx].BeginTime if tmpPerNum > 0 { tmpPerNum = uint64(printfList[idx].TotalNum) * 1000 / tmpPerNum } else { tmpPerNum = uint64(printfList[idx].TotalNum) } printfListStr += " \n" + printfList[idx].MsgName + ":" + strconv.FormatInt(int64(tmpTime), 10) + "(ms) | " + strconv.FormatInt(int64(tmpPerNum), 10) + "(req-ack/s) | " + strconv.Itoa(int(printfList[idx].TotalNum)) + "(total)" } util.DebugF("printfListStr=%v", printfListStr) } } func CreateAcceptor(param ServiceParam, serverConfig service.ConfigServerNode) rocommon.ServerNode { if param.ServiceType == "" { param.ServiceType = "tcpAcceptor" } node := socket.NewServerNode(param.ServiceType, param.ServiceName, param.LisAddr, Queue) //消息处理函数,根据给定的服务器类型来获取,例如gate,game,db,auth msgPrcFunc := serverproto.GetMessageHandler(param.ServiceName) //通过QueueEventCall函数加入直接函数的回调队列中 socket.SetProcessorRPC(node, param.ProcName, func(e rocommon.ProcEvent) { if msgPrcFunc != nil { msgPrcFunc(e) //nowTime := util.GetTimeMilliseconds() //msgPrcFunc(e) //nowTime1 := util.GetTimeMilliseconds() //switch in := e.(type) { //case *model.RecvGateMsgEvent: // //tmpInfo := rocommon.MessageInfoByMsg(in.Msg()) // //if tmpInfo != nil && tmpInfo.ID == 1066 { // // delTime := nowTime1 - nowTime // // recvTime := nowTime - in.KvTime // // util.DebugF("kvtime cid=%v idx=%v msg=%v deltime=%v recvtime=%v", // // in.ClientID, idxMap[in.ClientID], tmpInfo.ID, delTime, recvTime) // // //if delTime > 0 { // // // util.DebugF("kvtime cid=%v idx=%v deltime=%v msg=%T", in.ClientID, idxMap[in.ClientID], delTime, tmpInfo.ID) // // //} // //} // tmpInfo := rocommon.MessageInfoByMsg(in.Msg()) // if tmpInfo != nil { // gateGameTime := nowTime - in.KvTime // gateGameProcTime := nowTime1 - in.KvTime // idxItem, ok := idxMap[tmpInfo.ID] // if ok { // idxItem.TotalNum++ // idxItem.TotalTime += gateGameProcTime + gateGameTime // idxItem.NowTime = nowTime1 // } else { // idxItem = &model.PerformKVTimeSt{ // TotalNum: 1, // TotalTime: gateGameProcTime + gateGameTime, // MsgName: tmpInfo.Type.Name(), // BeginTime: nowTime1, // NowTime: nowTime1, // AckMsgId: int32(tmpInfo.ID), // } // } // idxMap[tmpInfo.ID] = idxItem // KVTimeTest(nowTime1) // } //} } }) if opt, ok := node.(rocommon.TCPSocketOption); ok { opt.SetSocketBuff(40960*1024, 40960*1024, true) } property := node.(rocommon.ServerNodeProperty) //服务器类型节点,区号,区号中的当前编号 property.SetServerType(serverConfig.Node.Type) property.SetZone(serverConfig.Node.Zone) property.SetIndex(serverConfig.Node.Id) //session uuid 创建时的key node.(rocommon.SessionMagExport).SetUuidCreateKey(serverConfig.Node.Id) node.Start() //注册到服务器发现etcd中 service.ETCDRegister(node) //todo...添加到总的服务器管理中 return node } func CreateConnector(param ServiceParam) rocommon.ServerNode { if param.ServiceType == "" { param.ServiceType = "tcpConnector" } //通过服务器发现ETCD来实现连接 service.DiscoveryService(param.DiscoveryServiceName, param.DiscoveryServiceZone, func(mn service.MultiServerNode, sd *service.ETCDServiceDesc) { //不连接自己 serviceCfg := service.GetServiceConfig() if sd.Type == serviceCfg.Node.Type && sd.Zone == serviceCfg.Node.Zone && sd.Index == serviceCfg.Node.Id { return } node := socket.NewServerNode(param.ServiceType, param.ServiceName, sd.Host, Queue) msgPrcFunc := serverproto.GetMessageHandler(param.ServiceName) socket.SetProcessorRPC(node, param.ProcName, func(e rocommon.ProcEvent) { if msgPrcFunc != nil { msgPrcFunc(e) //nowTime := util.GetTimeMilliseconds() //msgPrcFunc(e) //delTime := util.GetTimeMilliseconds() - nowTime //if delTime > 5 { // log.Printf("deltime=%v msg=%T", util.GetTimeMilliseconds()-nowTime, e.(rocommon.ProcEvent).Msg()) //} } }) if opt, ok := node.(rocommon.TCPSocketOption); ok { opt.SetSocketBuff(40960*1024, 40960*1024, true) //15s无读写断开,服务器之间已经添加心跳来位置读写(非调试模式启动) if !service.DebugMode { opt.SetSocketDeadline(time.Second*15, time.Second*15) } } //通过etcd服务器发现来处理,如果发现重复的节点会解除当前正在重连或者执行的节点 node.(rocommon.TCPConnector).SetReconnectTime(3 * time.Second) //如果连接的是服务器的节点那么需要加入当前本节点的信息,用来处理服务器节点之间的身份验证 // ServerTCPEventHook中会使用 node.(rocommon.ContextSet).SetContextData("sid", sd, "sid") //添加到服务器发现管理器中 mn.AddNode(sd, node) //util.DebugF("NewServerNode:%v", node.TypeOfName()) node.Start() //todo...添加到总的服务器管理中 }) if param.ServiceName != model.SERVICE_NODE_TYPE_CROSSROUTER_STR && param.ServiceName != model.SERVICE_NODE_TYPE_CROSSSERVER_STR { service.InitServiceStartupTime(param.DiscoveryServiceZone) } return nil } //gate监听客户端处理 func CreateClientAcceptor(param ServiceParam, serverConfig service.ConfigServerNode) rocommon.ServerNode { if param.ServiceType == "" { param.ServiceType = "tcpAcceptor" } //不需要要消息处理队列 //node := socket.NewServerNode(param.ServiceType, param.ServiceName, param.LisAddr, nil) node := socket.NewServerNode(param.ServiceType, param.ServiceName, param.LisAddr, Queue) //没有具体的消息处理逻辑 //socket.SetProcessorRPC(node, param.ProcName, nil) msgPrcFunc := serverproto.GetMessageHandler(param.ServiceName) socket.SetProcessorRPC(node, param.ProcName, func(e rocommon.ProcEvent) { if msgPrcFunc != nil { msgPrcFunc(e) } }) if opt, ok := node.(rocommon.TCPSocketOption); ok { opt.SetSocketBuff(40960, 40960, true) //40秒无读,30秒无写断开 如果没有心跳了超时直接断开,调试期间可以不加 // 通过该方法来模拟心跳保持连接 if serverConfig.Node.Reconnect > 0 { opt.SetSocketDeadline(time.Second*40, time.Second*40) //读/写协程没有过滤超时事件,发生了操时操作就断开连接 } } property := node.(rocommon.ServerNodeProperty) //服务器类型节点,区号,区号中的当前编号 property.SetServerType(serverConfig.Node.Type) property.SetZone(serverConfig.Node.Zone) property.SetIndex(serverConfig.Node.Id) //session uuid 创建时的key node.(rocommon.SessionMagExport).SetUuidCreateKey(serverConfig.Node.Id) node.Start() //外层通过该对象来获取监听器上接收的链接 model.ClientSessionMag = node.(socket.SessionManager) //注册到服务器发现etcd中 service.ETCDRegister(node) //todo...添加到总的服务器管理中 return node } func CreateHttpConnector(param ServiceParam) rocommon.ServerNode { if param.ServiceType == "" { param.ServiceType = "httpConnector" } node := socket.NewServerNode(param.ServiceType, param.ServiceName, param.LisAddr, nil) return node } //func CreateHttpAcceptor(param ServiceParam, serverConfig service.ConfigServerNode) rocommon.ServerNode { // if param.ServiceType == "" { // param.ServiceType = "httpAcceptor" // } // node := socket.NewServerNode(param.ServiceType, param.ServiceName, param.LisAddr, Queue) // // msgPrcFunc := serverproto.GetMessageHandler(param.ServiceName) // socket.SetProcessorRPC(node, param.ProcName, func(e rocommon.ProcEvent) { // if matcher, ok := e.Session().(http.RequestProc); ok { // msg.WebGMMsgProcess(matcher, e) // } else { // msgPrcFunc(e) // } // }) // // node.Start() // // return node //} //gate监听客户端处理 func CreateWebSocketAcceptor(param ServiceParam, serverConfig service.ConfigServerNode) rocommon.ServerNode { if param.ServiceType == "" { param.ServiceType = "wsAcceptor" } //不需要要消息处理队列 //node := socket.NewServerNode(param.ServiceType, param.ServiceName, param.LisAddr, nil) node := socket.NewServerNode(param.ServiceType, param.ServiceName, param.LisAddr, Queue) //没有具体的消息处理逻辑 //socket.SetProcessorRPC(node, param.ProcName, nil) msgPrcFunc := serverproto.GetMessageHandler(param.ServiceName) socket.SetProcessorRPC(node, param.ProcName, func(e rocommon.ProcEvent) { if msgPrcFunc != nil { msgPrcFunc(e) } }) if opt, ok := node.(rocommon.TCPSocketOption); ok { opt.SetSocketBuff(11264, 11264, true) //40秒无读,30秒无写断开 如果没有心跳了超时直接断开,调试期间可以不加 // 通过该方法来模拟心跳保持连接 if serverConfig.Node.Reconnect > 0 { opt.SetSocketDeadline(time.Second*40, time.Second*40) //读/写协程没有过滤超时事件,发生了操时操作就断开连接 } } property := node.(rocommon.ServerNodeProperty) //服务器类型节点,区号,区号中的当前编号 property.SetServerType(serverConfig.Node.Type) property.SetZone(serverConfig.Node.Zone) property.SetIndex(serverConfig.Node.Id) //session uuid 创建时的key node.(rocommon.SessionMagExport).SetUuidCreateKey(serverConfig.Node.Id) node.Start() //外层通过该对象来获取监听器上接收的链接 model.ClientSessionMag = node.(socket.SessionManager) //注册到服务器发现etcd中 service.ETCDRegister(node) //todo...添加到总的服务器管理中 return node }