完善文档

node/service/discovery.go

@@ -4,7 +4,9 @@ import (
 	"context"
 	"sort"
 	"strings"
+	"sync"
 	"sync/atomic"
+	"time"
 
 	"hk4e/common/region"
 	"hk4e/node/api"
@@ -35,28 +37,29 @@ type ServerInstance struct {
 	appId            string
 	gateServerIpAddr string
 	gateServerPort   uint32
+	version          string
+	lastAliveTime    int64
 }
 
 type DiscoveryService struct {
-	regionEc2b *random.Ec2b
-	// TODO 加锁
-	serverInstanceMap map[string]map[string]*ServerInstance
-	serverAppIdMap    map[string]bool
-	gsIdCounter       uint32
+	regionEc2b        *random.Ec2b         // 全局区服密钥信息
+	serverInstanceMap map[string]*sync.Map // 全部服务器实例集合 key:服务器类型 value:服务器实例集合 -> key:appid value:服务器实例
+	serverAppIdMap    map[string]bool      // 服务器appid集合 key:appid value:是否存在
+	gsIdCounter       uint32               // GSID计数器
 }
 
 func NewDiscoveryService() *DiscoveryService {
 	r := new(DiscoveryService)
 	r.regionEc2b = region.NewRegionEc2b()
 	logger.Info("region ec2b create ok, seed: %v", r.regionEc2b.Seed())
-	r.serverInstanceMap = map[string]map[string]*ServerInstance{
-		api.GATE:        make(map[string]*ServerInstance),
-		api.GS:          make(map[string]*ServerInstance),
-		api.FIGHT:       make(map[string]*ServerInstance),
-		api.PATHFINDING: make(map[string]*ServerInstance),
-	}
+	r.serverInstanceMap = make(map[string]*sync.Map)
+	r.serverInstanceMap[api.GATE] = &sync.Map{}
+	r.serverInstanceMap[api.GS] = &sync.Map{}
+	r.serverInstanceMap[api.FIGHT] = &sync.Map{}
+	r.serverInstanceMap[api.PATHFINDING] = &sync.Map{}
 	r.serverAppIdMap = make(map[string]bool)
 	r.gsIdCounter = 0
+	go r.removeDeadServer()
 	return r
 }
 
@@ -84,8 +87,9 @@ func (s *DiscoveryService) RegisterServer(ctx context.Context, req *api.Register
 		logger.Info("register new gate server, ip: %v, port: %v", req.GateServerAddr.IpAddr, req.GateServerAddr.Port)
 		inst.gateServerIpAddr = req.GateServerAddr.IpAddr
 		inst.gateServerPort = req.GateServerAddr.Port
+		inst.version = req.Version
 	}
-	instMap[appId] = inst
+	instMap.Store(appId, inst)
 	logger.Info("new server appid is: %v", appId)
 	rsp := &api.RegisterServerRsp{
 		AppId: appId,
@@ -103,11 +107,11 @@ func (s *DiscoveryService) CancelServer(ctx context.Context, req *api.CancelServ
 	if !exist {
 		return nil, errors.New("server type not exist")
 	}
-	_, exist = instMap[req.AppId]
+	_, exist = instMap.Load(req.AppId)
 	if !exist {
 		return nil, errors.New("server not exist")
 	}
-	delete(instMap, req.AppId)
+	instMap.Delete(req.AppId)
 	return &api.NullMsg{}, nil
 }
 
@@ -117,12 +121,12 @@ func (s *DiscoveryService) KeepaliveServer(ctx context.Context, req *api.Keepali
 	if !exist {
 		return nil, errors.New("server type not exist")
 	}
-	inst, exist := instMap[req.AppId]
+	inst, exist := instMap.Load(req.AppId)
 	if !exist {
 		return nil, errors.New("server not exist")
 	}
-	// TODO
-	_ = inst
+	serverInstance := inst.(*ServerInstance)
+	serverInstance.lastAliveTime = time.Now().Unix()
 	return &api.NullMsg{}, nil
 }
 
@@ -133,7 +137,7 @@ func (s *DiscoveryService) GetServerAppId(ctx context.Context, req *api.GetServe
 	if !exist {
 		return nil, errors.New("server type not exist")
 	}
-	if len(instMap) == 0 {
+	if s.getServerInstanceMapLen(instMap) == 0 {
 		return nil, errors.New("no server found")
 	}
 	inst := s.getRandomServerInstance(instMap)
@@ -152,16 +156,28 @@ func (s *DiscoveryService) GetRegionEc2B(ctx context.Context, req *api.NullMsg)
 }
 
 // GetGateServerAddr 获取负载最小的网关服务器的地址和端口
-func (s *DiscoveryService) GetGateServerAddr(ctx context.Context, req *api.NullMsg) (*api.GateServerAddr, error) {
+func (s *DiscoveryService) GetGateServerAddr(ctx context.Context, req *api.GetGateServerAddrReq) (*api.GateServerAddr, error) {
 	logger.Debug("get gate server addr")
 	instMap, exist := s.serverInstanceMap[api.GATE]
 	if !exist {
 		return nil, errors.New("gate server not exist")
 	}
-	if len(instMap) == 0 {
+	if s.getServerInstanceMapLen(instMap) == 0 {
 		return nil, errors.New("no gate server found")
 	}
-	inst := s.getRandomServerInstance(instMap)
+	versionInstMap := sync.Map{}
+	instMap.Range(func(key, value any) bool {
+		serverInstance := value.(*ServerInstance)
+		if serverInstance.version != req.Version {
+			return true
+		}
+		versionInstMap.Store(key, serverInstance)
+		return true
+	})
+	if s.getServerInstanceMapLen(&versionInstMap) == 0 {
+		return nil, errors.New("no gate server found")
+	}
+	inst := s.getRandomServerInstance(&versionInstMap)
 	logger.Debug("get gate server addr is, ip: %v, port: %v", inst.gateServerIpAddr, inst.gateServerPort)
 	return &api.GateServerAddr{
 		IpAddr: inst.gateServerIpAddr,
@@ -169,13 +185,42 @@ func (s *DiscoveryService) GetGateServerAddr(ctx context.Context, req *api.NullM
 	}, nil
 }
 
-func (s *DiscoveryService) getRandomServerInstance(instMap map[string]*ServerInstance) *ServerInstance {
+func (s *DiscoveryService) getRandomServerInstance(instMap *sync.Map) *ServerInstance {
 	instList := make(ServerInstanceSortList, 0)
-	for _, v := range instMap {
-		instList = append(instList, v)
-	}
+	instMap.Range(func(key, value any) bool {
+		instList = append(instList, value.(*ServerInstance))
+		return true
+	})
 	sort.Stable(instList)
 	index := random.GetRandomInt32(0, int32(len(instList)-1))
 	inst := instList[index]
 	return inst
 }
+
+func (s *DiscoveryService) getServerInstanceMapLen(instMap *sync.Map) int {
+	count := 0
+	instMap.Range(func(key, value any) bool {
+		count++
+		return true
+	})
+	return count
+}
+
+// 定时移除掉线服务器
+func (s *DiscoveryService) removeDeadServer() {
+	ticker := time.NewTicker(time.Second * 60)
+	for {
+		<-ticker.C
+		nowTime := time.Now().Unix()
+		for _, instMap := range s.serverInstanceMap {
+			instMap.Range(func(key, value any) bool {
+				serverInstance := value.(*ServerInstance)
+				if nowTime-serverInstance.lastAliveTime > 60 {
+					logger.Warn("remove dead server, server type: %v, appid: %v", serverInstance.serverType, serverInstance.appId)
+					instMap.Delete(key)
+				}
+				return true
+			})
+		}
+	}
+}