Ma/origin
1
0
Fork 0
mirror of https://github.com/duanhf2012/origin.git synced 2026-05-21 09:07:30 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

整理优化cluster

Browse Source
This commit is contained in:
orgin
2022-07-11 10:55:57 +08:00
parent 769f680b17
commit 3763f7d848
1 changed files with 8 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files

22
cluster/cluster.go
View File

@@ -46,10 +46,11 @@ type Cluster struct {
globalCfg interface{} //全局配置 globalCfg interface{} //全局配置
localServiceCfg map[string]interface{} //map[serviceName]配置数据* localServiceCfg map[string]interface{} //map[serviceName]配置数据*
mapRpc map[int]NodeRpcInfo //nodeId
serviceDiscovery IServiceDiscovery //服务发现接口 serviceDiscovery IServiceDiscovery //服务发现接口
locker sync.RWMutex //结点与服务关系保护锁 locker sync.RWMutex //结点与服务关系保护锁
mapRpc map[int]NodeRpcInfo //nodeId
mapIdNode map[int]NodeInfo //map[NodeId]NodeInfo mapIdNode map[int]NodeInfo //map[NodeId]NodeInfo
mapServiceNode map[string]map[int]struct{} //map[serviceName]map[NodeId] mapServiceNode map[string]map[int]struct{} //map[serviceName]map[NodeId]
@@ -95,9 +96,10 @@ func (cls *Cluster) DelNode(nodeId int, immediately bool) {
return return
} }
cls.locker.Lock() cls.locker.Lock()
defer cls.locker.Unlock()
nodeInfo, ok := cls.mapIdNode[nodeId] nodeInfo, ok := cls.mapIdNode[nodeId]
if ok == false { if ok == false {
cls.locker.Unlock()
return return
} }
@@ -113,7 +115,6 @@ func (cls *Cluster) DelNode(nodeId int, immediately bool) {
if rpc.client.IsConnected() { if rpc.client.IsConnected() {
nodeInfo.status = Discard nodeInfo.status = Discard
rpc.client.Unlock() rpc.client.Unlock()
cls.locker.Unlock()
log.SRelease("Discard node ", nodeInfo.NodeId, " ", nodeInfo.ListenAddr) log.SRelease("Discard node ", nodeInfo.NodeId, " ", nodeInfo.ListenAddr)
return return
} }
@@ -127,7 +128,6 @@ func (cls *Cluster) DelNode(nodeId int, immediately bool) {
delete(cls.mapIdNode, nodeId) delete(cls.mapIdNode, nodeId)
delete(cls.mapRpc, nodeId) delete(cls.mapRpc, nodeId)
cls.locker.Unlock()
if ok == true { if ok == true {
rpc.client.Close(false) rpc.client.Close(false)
} }
@@ -368,6 +368,7 @@ func (cls *Cluster) triggerRpcEvent(bConnect bool, clientSeq uint32, nodeId int)
cls.locker.Unlock() cls.locker.Unlock()
cls.rpcEventLocker.Lock() cls.rpcEventLocker.Lock()
defer cls.rpcEventLocker.Unlock()
for serviceName, _ := range cls.mapServiceListenRpcEvent { for serviceName, _ := range cls.mapServiceListenRpcEvent {
ser := service.GetService(serviceName) ser := service.GetService(serviceName)
if ser == nil { if ser == nil {
@@ -380,7 +381,6 @@ func (cls *Cluster) triggerRpcEvent(bConnect bool, clientSeq uint32, nodeId int)
eventData.NodeId = nodeId eventData.NodeId = nodeId
ser.(service.IModule).NotifyEvent(&eventData) ser.(service.IModule).NotifyEvent(&eventData)
} }
cls.rpcEventLocker.Unlock()
} }
@@ -441,13 +441,7 @@ func (cls *Cluster) UnReDiscoveryEvent(serviceName string) {
cls.rpcEventLocker.Unlock() cls.rpcEventLocker.Unlock()
} }
func (cls *Cluster) FetchAllNodeId(fetchFun func(nodeId int)) {
cls.locker.Lock()
for nodeId, _ := range cls.mapIdNode {
fetchFun(nodeId)
}
cls.locker.Unlock()
}
func HasService(nodeId int, serviceName string) bool { func HasService(nodeId int, serviceName string) bool {
cluster.locker.RLock() cluster.locker.RLock()
@@ -485,8 +479,8 @@ func (cls *Cluster) GetGlobalCfg() interface{} {
func (cls *Cluster) GetNodeInfo(nodeId int) (NodeInfo,bool) { func (cls *Cluster) GetNodeInfo(nodeId int) (NodeInfo,bool) {
cls.locker.Lock() cls.locker.RLock()
defer cls.locker.Unlock() defer cls.locker.RUnlock()
nodeInfo,ok:= cls.mapIdNode[nodeId] nodeInfo,ok:= cls.mapIdNode[nodeId]
return nodeInfo,ok return nodeInfo,ok