RoServer/RO_Server_Trunk-branch_bt/roserver/baseserver/init.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
}