373 lines
12 KiB
Go
373 lines
12 KiB
Go
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
|
||
}
|