优化RankService

导出RankData.RefreshTimestamp

README.md

@@ -19,7 +19,7 @@ Hello world!
 ```go
 go get -v -u  github.com/duanhf2012/origin
 ```
-
+[README.md](README.md)
 于是下载到GOPATH环境目录中,在src中加入main.go,内容如下：
 
 ```go
@@ -858,8 +858,7 @@ origin引擎默认使用读取所有结点配置的进行确认结点有哪些Se
 	"MasterDiscoveryNode": [{
 		"NodeId": 2,
 		"ListenAddr": "127.0.0.1:10001",
-		"MaxRpcParamLen": 409600,
-		"NeighborService":["HttpGateService"]
+		"MaxRpcParamLen": 409600
 	},
 	{
 		"NodeId": 1,
@@ -876,17 +875,18 @@ origin引擎默认使用读取所有结点配置的进行确认结点有哪些Se
 		"Private": false,
 		"remark": "//以_打头的，表示只在本机进程，不对整个子网开发",
 		"ServiceList": ["_TestService1", "TestService9", "TestService10"],
-		"DiscoveryService": ["TestService8"]
+	    "MasterDiscoveryService": [
+		    {
+                "MasterNodeId": 2,
+                "DiscoveryService": ["TestService8"]
+            }
+	    ]
 	}]
 }
 ```
+MasterDiscoveryNode: 配置了结点Id为1的服务发现Master，他的监听地址ListenAddr为127.0.0.1:8801，结点为2的也是一个服务发现Master。NodeId为1的结点会从结点为1和2的网络中发现服务。
 
-新上有两新不同的字段分别为MasterDiscoveryNode与DiscoveryService。其中:
-
-MasterDiscoveryNode中配置了结点Id为1的服务发现Master，他的监听地址ListenAddr为127.0.0.1:8801，结点为2的也是一个服务发现Master，不同在于多了"NeighborService":["HttpGateService"]配置。如果"NeighborService"有配置具体的服务时，则表示该结点是一个邻居Master结点。当前运行的Node结点会从该Master结点上筛选HttpGateService的服务，并且当前运行的Node结点不会向上同步本地所有公开的服务，和邻居结点关系是单向的。
-
-NeighborService可以用在当有多个以Master中心结点的网络，发现跨网络的服务场景。
-DiscoveryService表示将筛选origin网络中的TestService8服务，注意如果DiscoveryService不配置，则筛选功能不生效。
+MasterDiscoveryService: 表示将筛选origin网络中MasterNodeId为2中的TestService8服务，注意如果MasterDiscoveryService不配置，则筛选功能不生效。MasterNodeId也可以填为0，表示NodeId为1的结点，在所有网络中只发现TestService8的服务。
 
 第八章：HttpService使用
 -----------------------
@@ -1062,7 +1062,7 @@ func (slf *TestTcpService) OnRequest (clientid uint64,msg proto.Message){
 提交bug及特性: https://github.com/duanhf2012/origin/issues
 
 [因服务器是由个人维护，如果这个项目对您有帮助，您可以点我进行捐赠，感谢！](http://www.cppblog.com/images/cppblog_com/API/21416/r_pay.jpg "Thanks!")
-
+![](http://www.cppblog.com/images/cppblog_com/API/21416/r_pay.jpg)
 特别感谢以下赞助网友：
 
 ```
@@ -1071,4 +1071,5 @@ _
 死磕代码
 bp-li
 阿正
+大头
 ```

cluster/cluster.go

@@ -36,6 +36,7 @@ type NodeInfo struct {
 	PublicServiceList []string //对外公开的服务列表
 	MasterDiscoveryService  []MasterDiscoveryService //筛选发现的服务，如果不配置，不进行筛选
 	status            NodeStatus
+	Retire bool
 }
 
 type NodeRpcInfo struct {
@@ -55,8 +56,8 @@ type Cluster struct {
 
 
 	locker         sync.RWMutex                //结点与服务关系保护锁
-	mapRpc           map[int]NodeRpcInfo    //nodeId
-	mapIdNode      map[int]NodeInfo            //map[NodeId]NodeInfo
+	mapRpc           map[int]*NodeRpcInfo    //nodeId
+	//mapIdNode      map[int]NodeInfo            //map[NodeId]NodeInfo
 	mapServiceNode map[string]map[int]struct{} //map[serviceName]map[NodeId]
 
 	rpcServer                rpc.Server
@@ -87,10 +88,11 @@ func (cls *Cluster) Stop() {
 
 func (cls *Cluster) DiscardNode(nodeId int) {
 	cls.locker.Lock()
-	nodeInfo, ok := cls.mapIdNode[nodeId]
+	nodeInfo, ok := cls.mapRpc[nodeId]
+	bDel := (ok == true) &&  nodeInfo.nodeInfo.status == Discard
 	cls.locker.Unlock()
 
-	if ok == true && nodeInfo.status == Discard {
+	if bDel {
 		cls.DelNode(nodeId, true)
 	}
 }
@@ -103,39 +105,30 @@ func (cls *Cluster) DelNode(nodeId int, immediately bool) {
 	cls.locker.Lock()
 	defer cls.locker.Unlock()
 
-	nodeInfo, ok := cls.mapIdNode[nodeId]
+	rpc, ok := cls.mapRpc[nodeId]
 	if ok == false {
 		return
 	}
 
-	rpc, ok := cls.mapRpc[nodeId]
-	for {
-		//立即删除
-		if immediately || ok == false {
-			break
-		}
-
+	if immediately ==false {
 		//正在连接中不主动断开，只断开没有连接中的
 		if rpc.client.IsConnected() {
-			nodeInfo.status = Discard
-			log.Info("Discard node",log.Int("nodeId",nodeInfo.NodeId),log.String("ListenAddr", nodeInfo.ListenAddr))
+			rpc.nodeInfo.status = Discard
+			log.Info("Discard node",log.Int("nodeId",rpc.nodeInfo.NodeId),log.String("ListenAddr", rpc.nodeInfo.ListenAddr))
 			return
 		}
-
-		break
 	}
 
-	for _, serviceName := range nodeInfo.ServiceList {
+	for _, serviceName := range rpc.nodeInfo.ServiceList {
 		cls.delServiceNode(serviceName, nodeId)
 	}
 
-	delete(cls.mapIdNode, nodeId)
 	delete(cls.mapRpc, nodeId)
 	if ok == true {
 		rpc.client.Close(false)
 	}
 
-	log.Info("remove node ",log.Int("NodeId", nodeInfo.NodeId),log.String("ListenAddr", nodeInfo.ListenAddr))
+	log.Info("remove node ",log.Int("NodeId", rpc.nodeInfo.NodeId),log.String("ListenAddr", rpc.nodeInfo.ListenAddr))
 }
 
 func (cls *Cluster) serviceDiscoveryDelNode(nodeId int, immediately bool) {
@@ -168,9 +161,9 @@ func (cls *Cluster) serviceDiscoverySetNodeInfo(nodeInfo *NodeInfo) {
 	defer cls.locker.Unlock()
 
 	//先清一次的NodeId对应的所有服务清理
-	lastNodeInfo, ok := cls.mapIdNode[nodeInfo.NodeId]
+	lastNodeInfo, ok := cls.mapRpc[nodeInfo.NodeId]
 	if ok == true {
-		for _, serviceName := range lastNodeInfo.ServiceList {
+		for _, serviceName := range lastNodeInfo.nodeInfo.ServiceList {
 			cls.delServiceNode(serviceName, nodeInfo.NodeId)
 		}
 	}
@@ -189,27 +182,22 @@ func (cls *Cluster) serviceDiscoverySetNodeInfo(nodeInfo *NodeInfo) {
 		}
 		cls.mapServiceNode[serviceName][nodeInfo.NodeId] = struct{}{}
 	}
-	cls.mapIdNode[nodeInfo.NodeId] = *nodeInfo
-	log.Info("Discovery nodeId",log.Int("NodeId", nodeInfo.NodeId),log.Any("services:", nodeInfo.PublicServiceList))
 
-	//已经存在连接，则不需要进行设置
-	if _, rpcInfoOK := cls.mapRpc[nodeInfo.NodeId]; rpcInfoOK == true {
+	if lastNodeInfo != nil {
+		log.Info("Discovery nodeId",log.Int("NodeId", nodeInfo.NodeId),log.Any("services:", nodeInfo.PublicServiceList),log.Bool("Retire",nodeInfo.Retire))
+		lastNodeInfo.nodeInfo = *nodeInfo
 		return
 	}
 
+	//不存在时，则建立连接
 	rpcInfo := NodeRpcInfo{}
 	rpcInfo.nodeInfo = *nodeInfo
 	rpcInfo.client =rpc.NewRClient(nodeInfo.NodeId, nodeInfo.ListenAddr, nodeInfo.MaxRpcParamLen,cls.localNodeInfo.CompressBytesLen,cls.triggerRpcEvent)
-	cls.mapRpc[nodeInfo.NodeId] = rpcInfo
+	cls.mapRpc[nodeInfo.NodeId] = &rpcInfo
+	log.Info("Discovery nodeId and new rpc client",log.Int("NodeId", nodeInfo.NodeId),log.Any("services:", nodeInfo.PublicServiceList),log.Bool("Retire",nodeInfo.Retire),log.String("nodeListenAddr",nodeInfo.ListenAddr))
 }
 
-func (cls *Cluster) buildLocalRpc() {
-	rpcInfo := NodeRpcInfo{}
-	rpcInfo.nodeInfo = cls.localNodeInfo
-	rpcInfo.client = rpc.NewLClient(rpcInfo.nodeInfo.NodeId)
 
-	cls.mapRpc[cls.localNodeInfo.NodeId] = rpcInfo
-}
 
 func (cls *Cluster) Init(localNodeId int, setupServiceFun SetupServiceFun) error {
 	//1.初始化配置
@@ -219,7 +207,6 @@ func (cls *Cluster) Init(localNodeId int, setupServiceFun SetupServiceFun) error
 	}
 
 	cls.rpcServer.Init(cls)
-	cls.buildLocalRpc()
 
 	//2.安装服务发现结点
 	cls.SetupServiceDiscovery(localNodeId, setupServiceFun)
@@ -313,27 +300,33 @@ func (cls *Cluster) FindRpcHandler(serviceName string) rpc.IRpcHandler {
 	return pService.GetRpcHandler()
 }
 
-func (cls *Cluster) getRpcClient(nodeId int) *rpc.Client {
+func (cls *Cluster) getRpcClient(nodeId int) (*rpc.Client,bool) {
 	c, ok := cls.mapRpc[nodeId]
 	if ok == false {
-		return nil
+		return nil,false
 	}
 
-	return c.client
+	return c.client,c.nodeInfo.Retire
 }
 
-func (cls *Cluster) GetRpcClient(nodeId int) *rpc.Client {
+func (cls *Cluster) GetRpcClient(nodeId int) (*rpc.Client,bool) {
 	cls.locker.RLock()
 	defer cls.locker.RUnlock()
 	return cls.getRpcClient(nodeId)
 }
 
-func GetRpcClient(nodeId int, serviceMethod string, clientList []*rpc.Client) (error, int) {
+func GetRpcClient(nodeId int, serviceMethod string,filterRetire bool, clientList []*rpc.Client) (error, int) {
 	if nodeId > 0 {
-		pClient := GetCluster().GetRpcClient(nodeId)
+		pClient,retire := GetCluster().GetRpcClient(nodeId)
 		if pClient == nil {
 			return fmt.Errorf("cannot find  nodeid %d!", nodeId), 0
 		}
+
+		//如果需要筛选掉退休结点
+		if filterRetire == true && retire == true {
+			return fmt.Errorf("cannot find  nodeid %d!", nodeId), 0
+		}
+
 		clientList[0] = pClient
 		return nil, 1
 	}
@@ -345,7 +338,7 @@ func GetRpcClient(nodeId int, serviceMethod string, clientList []*rpc.Client) (e
 	serviceName := serviceMethod[:findIndex]
 
 	//1.找到对应的rpcNodeid
-	return GetCluster().GetNodeIdByService(serviceName, clientList, true)
+	return GetCluster().GetNodeIdByService(serviceName, clientList, filterRetire)
 }
 
 func GetRpcServer() *rpc.Server {
@@ -353,10 +346,19 @@ func GetRpcServer() *rpc.Server {
 }
 
 func (cls *Cluster) IsNodeConnected(nodeId int) bool {
-	pClient := cls.GetRpcClient(nodeId)
+	pClient,_ := cls.GetRpcClient(nodeId)
 	return pClient != nil && pClient.IsConnected()
 }
 
+func (cls *Cluster) IsNodeRetire(nodeId int) bool {
+	cls.locker.RLock()
+	defer cls.locker.RUnlock()
+
+	_,retire :=cls.getRpcClient(nodeId)
+	return retire
+}
+
+
 func (cls *Cluster) triggerRpcEvent(bConnect bool, clientId uint32, nodeId int) {
 	cls.locker.Lock()
 	nodeInfo, ok := cls.mapRpc[nodeId]
@@ -475,11 +477,14 @@ func (cls *Cluster) GetGlobalCfg() interface{} {
 	return cls.globalCfg
 }
 
-
 func (cls *Cluster) GetNodeInfo(nodeId int) (NodeInfo,bool) {
 	cls.locker.RLock()
 	defer cls.locker.RUnlock()
 
-	nodeInfo,ok:= cls.mapIdNode[nodeId]
-	return nodeInfo,ok
+	nodeInfo,ok:= cls.mapRpc[nodeId]
+	if ok == false || nodeInfo == nil {
+		return NodeInfo{},false
+	}
+
+	return nodeInfo.nodeInfo,true
 }

cluster/dynamicdiscovery.go

@@ -7,6 +7,7 @@ import (
 	"github.com/duanhf2012/origin/service"
 	"time"
 	"github.com/duanhf2012/origin/util/timer"
+	"google.golang.org/protobuf/proto"
 )
 
 const DynamicDiscoveryMasterName = "DiscoveryMaster"
@@ -14,6 +15,7 @@ const DynamicDiscoveryClientName = "DiscoveryClient"
 const RegServiceDiscover = DynamicDiscoveryMasterName + ".RPC_RegServiceDiscover"
 const SubServiceDiscover = DynamicDiscoveryClientName + ".RPC_SubServiceDiscover"
 const AddSubServiceDiscover = DynamicDiscoveryMasterName + ".RPC_AddSubServiceDiscover"
+const NodeRetireRpcMethod = DynamicDiscoveryMasterName+".RPC_NodeRetire"
 
 type DynamicDiscoveryMaster struct {
 	service.Service
@@ -30,6 +32,7 @@ type DynamicDiscoveryClient struct {
 	localNodeId   int
 
 	mapDiscovery map[int32]map[int32]struct{} //map[masterNodeId]map[nodeId]struct{}
+	bRetire bool
 }
 
 var masterService DynamicDiscoveryMaster
@@ -49,16 +52,32 @@ func (ds *DynamicDiscoveryMaster) isRegNode(nodeId int32) bool {
 	return ok
 }
 
-func (ds *DynamicDiscoveryMaster) addNodeInfo(nodeInfo *rpc.NodeInfo) {
-	if len(nodeInfo.PublicServiceList) == 0 {
+func (ds *DynamicDiscoveryMaster) updateNodeInfo(nInfo *rpc.NodeInfo) {
+	if _,ok:= ds.mapNodeInfo[nInfo.NodeId];ok == false {
 		return
 	}
 
-	_, ok := ds.mapNodeInfo[nodeInfo.NodeId]
+	nodeInfo := proto.Clone(nInfo).(*rpc.NodeInfo)
+	for i:=0;i<len(ds.nodeInfo);i++ {
+		if ds.nodeInfo[i].NodeId == nodeInfo.NodeId {
+			ds.nodeInfo[i]  = nodeInfo
+			break
+		}
+	}
+}
+
+func (ds *DynamicDiscoveryMaster) addNodeInfo(nInfo *rpc.NodeInfo) {
+	if len(nInfo.PublicServiceList) == 0 {
+		return
+	}
+
+	_, ok := ds.mapNodeInfo[nInfo.NodeId]
 	if ok == true {
 		return
 	}
-	ds.mapNodeInfo[nodeInfo.NodeId] = struct{}{}
+	ds.mapNodeInfo[nInfo.NodeId] = struct{}{}
+
+	nodeInfo := proto.Clone(nInfo).(*rpc.NodeInfo)
 	ds.nodeInfo = append(ds.nodeInfo, nodeInfo)
 }
 
@@ -87,16 +106,14 @@ func (ds *DynamicDiscoveryMaster) OnInit() error {
 func (ds *DynamicDiscoveryMaster) OnStart() {
 	var nodeInfo rpc.NodeInfo
 	localNodeInfo := cluster.GetLocalNodeInfo()
-	if localNodeInfo.Private == true {
-		return
-	}
-
 	nodeInfo.NodeId = int32(localNodeInfo.NodeId)
 	nodeInfo.NodeName = localNodeInfo.NodeName
 	nodeInfo.ListenAddr = localNodeInfo.ListenAddr
 	nodeInfo.PublicServiceList = localNodeInfo.PublicServiceList
 	nodeInfo.MaxRpcParamLen = localNodeInfo.MaxRpcParamLen
-	
+	nodeInfo.Private = localNodeInfo.Private
+	nodeInfo.Retire = localNodeInfo.Retire
+
 	ds.addNodeInfo(&nodeInfo)
 }
 
@@ -138,6 +155,19 @@ func (ds *DynamicDiscoveryMaster) RpcCastGo(serviceMethod string, args interface
 	}
 }
 
+func (ds *DynamicDiscoveryMaster) RPC_NodeRetire(req *rpc.NodeRetireReq, res *rpc.Empty) error {
+	log.Info("node is retire",log.Int32("nodeId",req.NodeInfo.NodeId),log.Bool("retire",req.NodeInfo.Retire))
+
+	ds.updateNodeInfo(req.NodeInfo)
+
+	var notifyDiscover rpc.SubscribeDiscoverNotify
+	notifyDiscover.MasterNodeId = int32(cluster.GetLocalNodeInfo().NodeId)
+	notifyDiscover.NodeInfo = append(notifyDiscover.NodeInfo, req.NodeInfo)
+	ds.RpcCastGo(SubServiceDiscover, &notifyDiscover)
+
+	return nil
+}
+
 // 收到注册过来的结点
 func (ds *DynamicDiscoveryMaster) RPC_RegServiceDiscover(req *rpc.ServiceDiscoverReq, res *rpc.Empty) error {
 	if req.NodeInfo == nil {
@@ -165,6 +195,8 @@ func (ds *DynamicDiscoveryMaster) RPC_RegServiceDiscover(req *rpc.ServiceDiscove
 	nodeInfo.PublicServiceList = req.NodeInfo.PublicServiceList
 	nodeInfo.ListenAddr = req.NodeInfo.ListenAddr
 	nodeInfo.MaxRpcParamLen = req.NodeInfo.MaxRpcParamLen
+	nodeInfo.Retire = req.NodeInfo.Retire
+
 	//主动删除已经存在的结点,确保先断开，再连接
 	cluster.serviceDiscoveryDelNode(nodeInfo.NodeId, true)
 
@@ -269,6 +301,9 @@ func (dc *DynamicDiscoveryClient) RPC_SubServiceDiscover(req *rpc.SubscribeDisco
 				nInfo.NodeName = nodeInfo.NodeName
 				nInfo.ListenAddr = nodeInfo.ListenAddr
 				nInfo.MaxRpcParamLen = nodeInfo.MaxRpcParamLen
+				nInfo.Retire = nodeInfo.Retire
+				nInfo.Private = nodeInfo.Private
+
 				mapNodeInfo[nodeInfo.NodeId] = nInfo
 			}
 
@@ -278,7 +313,6 @@ func (dc *DynamicDiscoveryClient) RPC_SubServiceDiscover(req *rpc.SubscribeDisco
 
 	//如果为完整同步，则找出差异的结点
 	var willDelNodeId []int32
-	//如果不是邻居结点，则做筛选
 	if req.IsFull == true {
 		diffNode := dc.fullCompareDiffNode(req.MasterNodeId, mapNodeInfo)
 		if len(diffNode) > 0 {
@@ -293,8 +327,7 @@ func (dc *DynamicDiscoveryClient) RPC_SubServiceDiscover(req *rpc.SubscribeDisco
 
 	//删除不必要的结点
 	for _, nodeId := range willDelNodeId {
-		nodeInfo,_ := cluster.GetNodeInfo(int(nodeId))
-		cluster.TriggerDiscoveryEvent(false,int(nodeId),nodeInfo.PublicServiceList)
+		cluster.TriggerDiscoveryEvent(false,int(nodeId),nil)
 		dc.removeMasterNode(req.MasterNodeId, int32(nodeId))
 		if dc.findNodeId(nodeId) == false {
 			dc.funDelService(int(nodeId), false)
@@ -328,6 +361,29 @@ func (dc *DynamicDiscoveryClient) OnNodeConnected(nodeId int) {
 	dc.regServiceDiscover(nodeId)
 }
 
+func (dc *DynamicDiscoveryClient) OnRetire(){
+	dc.bRetire = true
+
+	masterNodeList := cluster.GetDiscoveryNodeList()
+	for i:=0;i<len(masterNodeList);i++{
+		var nodeRetireReq rpc.NodeRetireReq
+
+		nodeRetireReq.NodeInfo = &rpc.NodeInfo{}
+		nodeRetireReq.NodeInfo.NodeId = int32(cluster.localNodeInfo.NodeId)
+		nodeRetireReq.NodeInfo.NodeName = cluster.localNodeInfo.NodeName
+		nodeRetireReq.NodeInfo.ListenAddr = cluster.localNodeInfo.ListenAddr
+		nodeRetireReq.NodeInfo.MaxRpcParamLen = cluster.localNodeInfo.MaxRpcParamLen
+		nodeRetireReq.NodeInfo.PublicServiceList =  cluster.localNodeInfo.PublicServiceList
+		nodeRetireReq.NodeInfo.Retire = dc.bRetire
+		nodeRetireReq.NodeInfo.Private = cluster.localNodeInfo.Private
+
+		err := dc.GoNode(int(masterNodeList[i].NodeId),NodeRetireRpcMethod,&nodeRetireReq)
+		if err!= nil {
+			log.Error("call "+NodeRetireRpcMethod+" is fail",log.ErrorAttr("err",err))
+		}
+	}
+}
+
 func (dc *DynamicDiscoveryClient) regServiceDiscover(nodeId int){
 	nodeInfo := cluster.GetMasterDiscoveryNodeInfo(nodeId)
 	if nodeInfo == nil {
@@ -341,7 +397,8 @@ func (dc *DynamicDiscoveryClient) regServiceDiscover(nodeId int){
 	req.NodeInfo.ListenAddr = cluster.localNodeInfo.ListenAddr
 	req.NodeInfo.MaxRpcParamLen = cluster.localNodeInfo.MaxRpcParamLen
 	req.NodeInfo.PublicServiceList =  cluster.localNodeInfo.PublicServiceList
-
+	req.NodeInfo.Retire = dc.bRetire
+	req.NodeInfo.Private = cluster.localNodeInfo.Private
 
 	//向Master服务同步本Node服务信息
 	err := dc.AsyncCallNode(nodeId, RegServiceDiscover, &req, func(res *rpc.Empty, err error) {
@@ -403,6 +460,9 @@ func (dc *DynamicDiscoveryClient) setNodeInfo(masterNodeId int32,nodeInfo *rpc.N
 	nInfo.NodeName = nodeInfo.NodeName
 	nInfo.ListenAddr = nodeInfo.ListenAddr
 	nInfo.MaxRpcParamLen = nodeInfo.MaxRpcParamLen
+	nInfo.Retire = nodeInfo.Retire
+	nInfo.Private = nodeInfo.Private
+
 	dc.funSetService(&nInfo)
 
 	return true

cluster/parsecfg.go

@@ -199,7 +199,11 @@ func (cls *Cluster) readLocalService(localNodeId int) error {
 }
 
 func (cls *Cluster) parseLocalCfg() {
-	cls.mapIdNode[cls.localNodeInfo.NodeId] = cls.localNodeInfo
+	rpcInfo := NodeRpcInfo{}
+	rpcInfo.nodeInfo = cls.localNodeInfo
+	rpcInfo.client = rpc.NewLClient(rpcInfo.nodeInfo.NodeId)
+
+	cls.mapRpc[cls.localNodeInfo.NodeId] = &rpcInfo
 
 	for _, sName := range cls.localNodeInfo.ServiceList {
 		if _, ok := cls.mapServiceNode[sName]; ok == false {
@@ -225,8 +229,7 @@ func (cls *Cluster) checkDiscoveryNodeList(discoverMasterNode []NodeInfo) bool {
 
 func (cls *Cluster) InitCfg(localNodeId int) error {
 	cls.localServiceCfg = map[string]interface{}{}
-	cls.mapRpc = map[int]NodeRpcInfo{}
-	cls.mapIdNode = map[int]NodeInfo{}
+	cls.mapRpc = map[int]*NodeRpcInfo{}
 	cls.mapServiceNode = map[string]map[int]struct{}{}
 
 	//加载本地结点的NodeList配置
@@ -263,17 +266,24 @@ func (cls *Cluster) IsConfigService(serviceName string) bool {
 	return ok
 }
 
-func (cls *Cluster) GetNodeIdByService(serviceName string, rpcClientList []*rpc.Client, bAll bool) (error, int) {
+
+func (cls *Cluster) GetNodeIdByService(serviceName string, rpcClientList []*rpc.Client, filterRetire bool) (error, int) {
 	cls.locker.RLock()
 	defer cls.locker.RUnlock()
 	mapNodeId, ok := cls.mapServiceNode[serviceName]
 	count := 0
 	if ok == true {
 		for nodeId, _ := range mapNodeId {
-			pClient := GetCluster().getRpcClient(nodeId)
-			if pClient == nil || (bAll == false && pClient.IsConnected() == false) {
+			pClient,retire := GetCluster().getRpcClient(nodeId)
+			if pClient == nil || pClient.IsConnected() == false {
 				continue
 			}
+
+			//如果需要筛选掉退休的，对retire状态的结点略过
+			if filterRetire == true && retire == true {
+				continue
+			}
+
 			rpcClientList[count] = pClient
 			count++
 			if count >= cap(rpcClientList) {

concurrent/concurrent.go

@@ -5,6 +5,7 @@ import (
 	"runtime"
 
 	"github.com/duanhf2012/origin/log"
+	"sync/atomic"
 )
 
 const defaultMaxTaskChannelNum = 1000000
@@ -21,6 +22,7 @@ type Concurrent struct {
 
 	tasks     chan task
 	cbChannel chan func(error)
+	open int32
 }
 
 /*
@@ -33,6 +35,10 @@ func (c *Concurrent) OpenConcurrentByNumCPU(cpuNumMul float32) {
 }
 
 func (c *Concurrent) OpenConcurrent(minGoroutineNum int32, maxGoroutineNum int32, maxTaskChannelNum int) {
+	if atomic.AddInt32(&c.open,1) > 1 {
+		panic("repeated calls to OpenConcurrent are not allowed!")
+	}
+	
 	c.tasks = make(chan task, maxTaskChannelNum)
 	c.cbChannel = make(chan func(error), maxTaskChannelNum)
 

event/eventtype.go

@@ -7,16 +7,15 @@ const (
 	ServiceRpcRequestEvent 		EventType = -1
 	ServiceRpcResponseEvent  	EventType = -2
 
-	Sys_Event_Tcp         			EventType = -3
-	Sys_Event_Http_Event  			EventType = -4
-	Sys_Event_WebSocket       EventType = -5
-	Sys_Event_Node_Event      EventType = -6
-	Sys_Event_DiscoverService EventType = -7
-	Sys_Event_DiscardGoroutine EventType = -8
-	Sys_Event_QueueTaskFinish  EventType = -9
-	
-	Sys_Event_User_Define EventType = 1
-
+	Sys_Event_Tcp         		EventType = -3
+	Sys_Event_Http_Event  		EventType = -4
+	Sys_Event_WebSocket       	EventType = -5
+	Sys_Event_Node_Event      	EventType = -6
+	Sys_Event_DiscoverService 	EventType = -7
+	Sys_Event_DiscardGoroutine 	EventType = -8
+	Sys_Event_QueueTaskFinish  	EventType = -9
+	Sys_Event_Retire       	EventType = -10
 
+	Sys_Event_User_Define 		EventType = 1
 )
 

go.mod

@@ -8,24 +8,30 @@ require (
 	github.com/gorilla/websocket v1.5.0
 	github.com/json-iterator/go v1.1.12
 	github.com/pierrec/lz4/v4 v4.1.18
+	github.com/shirou/gopsutil v3.21.11+incompatible
 	go.mongodb.org/mongo-driver v1.9.1
 	google.golang.org/protobuf v1.31.0
 	gopkg.in/mgo.v2 v2.0.0-20190816093944-a6b53ec6cb22
 )
 
 require (
+	github.com/go-ole/go-ole v1.2.6 // indirect
 	github.com/go-stack/stack v1.8.0 // indirect
 	github.com/golang/snappy v0.0.1 // indirect
 	github.com/klauspost/compress v1.13.6 // indirect
 	github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421 // indirect
 	github.com/modern-go/reflect2 v1.0.2 // indirect
 	github.com/pkg/errors v0.9.1 // indirect
+	github.com/tklauser/go-sysconf v0.3.13 // indirect
+	github.com/tklauser/numcpus v0.7.0 // indirect
 	github.com/xdg-go/pbkdf2 v1.0.0 // indirect
 	github.com/xdg-go/scram v1.0.2 // indirect
 	github.com/xdg-go/stringprep v1.0.2 // indirect
 	github.com/youmark/pkcs8 v0.0.0-20181117223130-1be2e3e5546d // indirect
+	github.com/yusufpapurcu/wmi v1.2.4 // indirect
 	golang.org/x/crypto v0.1.0 // indirect
 	golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9 // indirect
+	golang.org/x/sys v0.15.0 // indirect
 	golang.org/x/text v0.4.0 // indirect
 	gopkg.in/yaml.v2 v2.4.0 // indirect
 )

go.sum

@@ -1,12 +1,12 @@
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/go-ole/go-ole v1.2.6 h1:/Fpf6oFPoeFik9ty7siob0G6Ke8QvQEuVcuChpwXzpY=
+github.com/go-ole/go-ole v1.2.6/go.mod h1:pprOEPIfldk/42T2oK7lQ4v4JSDwmV0As9GaiUsvbm0=
 github.com/go-sql-driver/mysql v1.6.0 h1:BCTh4TKNUYmOmMUcQ3IipzF5prigylS7XXjEkfCHuOE=
 github.com/go-sql-driver/mysql v1.6.0/go.mod h1:DCzpHaOWr8IXmIStZouvnhqoel9Qv2LBy8hT2VhHyBg=
 github.com/go-stack/stack v1.8.0 h1:5SgMzNM5HxrEjV0ww2lTmX6E2Izsfxas4+YHWRs3Lsk=
 github.com/go-stack/stack v1.8.0/go.mod h1:v0f6uXyyMGvRgIKkXu+yp6POWl0qKG85gN/melR3HDY=
-github.com/gogo/protobuf v1.3.2 h1:Ov1cvc58UF3b5XjBnZv7+opcTcQFZebYjWzi34vdm4Q=
-github.com/gogo/protobuf v1.3.2/go.mod h1:P1XiOD3dCwIKUDQYPy72D8LYyHL2YPYrpS2s69NZV8Q=
 github.com/golang/protobuf v1.5.0/go.mod h1:FsONVRAS9T7sI+LIUmWTfcYkHO4aIWwzhcaSAoJOfIk=
 github.com/golang/snappy v0.0.1 h1:Qgr9rKW7uDUkrbSmQeiDsGa8SjGyCOGtuasMWwvp2P4=
 github.com/golang/snappy v0.0.1/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q=
@@ -20,8 +20,6 @@ github.com/gorilla/websocket v1.5.0 h1:PPwGk2jz7EePpoHN/+ClbZu8SPxiqlu12wZP/3sWm
 github.com/gorilla/websocket v1.5.0/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
 github.com/json-iterator/go v1.1.12 h1:PV8peI4a0ysnczrg+LtxykD8LfKY9ML6u2jnxaEnrnM=
 github.com/json-iterator/go v1.1.12/go.mod h1:e30LSqwooZae/UwlEbR2852Gd8hjQvJoHmT4TnhNGBo=
-github.com/kisielk/errcheck v1.5.0/go.mod h1:pFxgyoBC7bSaBwPgfKdkLd5X25qrDl4LWUI2bnpBCr8=
-github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck=
 github.com/klauspost/compress v1.13.6 h1:P76CopJELS0TiO2mebmnzgWaajssP/EszplttgQxcgc=
 github.com/klauspost/compress v1.13.6/go.mod h1:/3/Vjq9QcHkK5uEr5lBEmyoZ1iFhe47etQ6QUkpK6sk=
 github.com/kr/pretty v0.1.0 h1:L/CwN0zerZDmRFUapSPitk6f+Q3+0za1rQkzVuMiMFI=
@@ -40,6 +38,8 @@ github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=
 github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/shirou/gopsutil v3.21.11+incompatible h1:+1+c1VGhc88SSonWP6foOcLhvnKlUeu/erjjvaPEYiI=
+github.com/shirou/gopsutil v3.21.11+incompatible/go.mod h1:5b4v6he4MtMOwMlS0TUMTu2PcXUg8+E1lC7eC3UO/RA=
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
 github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
 github.com/stretchr/testify v1.6.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
@@ -47,6 +47,10 @@ github.com/stretchr/testify v1.7.0 h1:nwc3DEeHmmLAfoZucVR881uASk0Mfjw8xYJ99tb5Cc
 github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
 github.com/tidwall/pretty v1.0.0 h1:HsD+QiTn7sK6flMKIvNmpqz1qrpP3Ps6jOKIKMooyg4=
 github.com/tidwall/pretty v1.0.0/go.mod h1:XNkn88O1ChpSDQmQeStsy+sBenx6DDtFZJxhVysOjyk=
+github.com/tklauser/go-sysconf v0.3.13 h1:GBUpcahXSpR2xN01jhkNAbTLRk2Yzgggk8IM08lq3r4=
+github.com/tklauser/go-sysconf v0.3.13/go.mod h1:zwleP4Q4OehZHGn4CYZDipCgg9usW5IJePewFCGVEa0=
+github.com/tklauser/numcpus v0.7.0 h1:yjuerZP127QG9m5Zh/mSO4wqurYil27tHrqwRoRjpr4=
+github.com/tklauser/numcpus v0.7.0/go.mod h1:bb6dMVcj8A42tSE7i32fsIUCbQNllK5iDguyOZRUzAY=
 github.com/xdg-go/pbkdf2 v1.0.0 h1:Su7DPu48wXMwC3bs7MCNG+z4FhcyEuz5dlvchbq0B0c=
 github.com/xdg-go/pbkdf2 v1.0.0/go.mod h1:jrpuAogTd400dnrH08LKmI/xc1MbPOebTwRqcT5RDeI=
 github.com/xdg-go/scram v1.0.2 h1:akYIkZ28e6A96dkWNJQu3nmCzH3YfwMPQExUYDaRv7w=
@@ -55,47 +59,34 @@ github.com/xdg-go/stringprep v1.0.2 h1:6iq84/ryjjeRmMJwxutI51F2GIPlP5BfTvXHeYjyh
 github.com/xdg-go/stringprep v1.0.2/go.mod h1:8F9zXuvzgwmyT5DUm4GUfZGDdT3W+LCvS6+da4O5kxM=
 github.com/youmark/pkcs8 v0.0.0-20181117223130-1be2e3e5546d h1:splanxYIlg+5LfHAM6xpdFEAYOk8iySO56hMFq6uLyA=
 github.com/youmark/pkcs8 v0.0.0-20181117223130-1be2e3e5546d/go.mod h1:rHwXgn7JulP+udvsHwJoVG1YGAP6VLg4y9I5dyZdqmA=
-github.com/yuin/goldmark v1.1.27/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
-github.com/yuin/goldmark v1.2.1/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
+github.com/yusufpapurcu/wmi v1.2.4 h1:zFUKzehAFReQwLys1b/iSMl+JQGSCSjtVqQn9bBrPo0=
+github.com/yusufpapurcu/wmi v1.2.4/go.mod h1:SBZ9tNy3G9/m5Oi98Zks0QjeHVDvuK0qfxQmPyzfmi0=
 go.mongodb.org/mongo-driver v1.9.1 h1:m078y9v7sBItkt1aaoe2YlvWEXcD263e1a4E1fBrJ1c=
 go.mongodb.org/mongo-driver v1.9.1/go.mod h1:0sQWfOeY63QTntERDJJ/0SuKK0T1uVSgKCuAROlKEPY=
 golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
-golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
-golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
 golang.org/x/crypto v0.0.0-20201216223049-8b5274cf687f/go.mod h1:jdWPYTVW3xRLrWPugEBEK3UY2ZEsg3UU495nc5E+M+I=
 golang.org/x/crypto v0.1.0 h1:MDRAIl0xIo9Io2xV565hzXHw3zVseKrJKodhohM5CjU=
 golang.org/x/crypto v0.1.0/go.mod h1:RecgLatLF4+eUMCP1PoPZQb+cVrJcOPbHkTkbkB9sbw=
-golang.org/x/mod v0.2.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
-golang.org/x/mod v0.3.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
 golang.org/x/net v0.0.0-20190311183353-d8887717615a/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
 golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
-golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
-golang.org/x/net v0.0.0-20200226121028-0de0cce0169b/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
-golang.org/x/net v0.0.0-20201021035429-f5854403a974/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU=
 golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9 h1:SQFwaSi55rU7vdNs9Yr0Z324VNlrF+0wMqRXT4St8ck=
 golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
-golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20190916202348-b4ddaad3f8a3/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20191026070338-33540a1f6037/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
-golang.org/x/sys v0.0.0-20200930185726-fdedc70b468f/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.15.0 h1:h48lPFYpsTvQJZF4EKyI4aLHaev3CxivZmv7yZig9pc=
+golang.org/x/sys v0.15.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
 golang.org/x/term v0.0.0-20201117132131-f5c789dd3221/go.mod h1:Nr5EML6q2oocZ2LXRh80K7BxOlk5/8JxuGnuhpl+muw=
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
-golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
 golang.org/x/text v0.3.5/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
 golang.org/x/text v0.4.0 h1:BrVqGRd7+k1DiOgtnFvAkoQEWQvBc25ouMJM6429SFg=
 golang.org/x/text v0.4.0/go.mod h1:mrYo+phRRbMaCq/xk9113O4dZlRixOauAjOtrjsXDZ8=
 golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
 golang.org/x/tools v0.0.0-20190531172133-b3315ee88b7d/go.mod h1:/rFqwRUd4F7ZHNgwSSTFct+R/Kf4OFW1sUzUTQQTgfc=
-golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
-golang.org/x/tools v0.0.0-20200619180055-7c47624df98f/go.mod h1:EkVYQZoAsY45+roYkvgYkIh4xh/qjgUK9TdY2XT94GE=
-golang.org/x/tools v0.0.0-20210106214847-113979e3529a/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA=
-golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
-golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
+golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543 h1:E7g+9GITq07hpfrRu66IVDexMakfv52eLZ2CXBWiKr4=
 golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
-golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1 h1:go1bK/D/BFZV2I8cIQd1NKEZ+0owSTG1fDTci4IqFcE=
-golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 google.golang.org/protobuf v1.26.0-rc.1/go.mod h1:jlhhOSvTdKEhbULTjvd4ARK9grFBp09yW+WbY/TyQbw=
 google.golang.org/protobuf v1.31.0 h1:g0LDEJHgrBl9N9r17Ru3sqWhkIx2NB67okBHPwC7hs8=
 google.golang.org/protobuf v1.31.0/go.mod h1:HV8QOd/L58Z+nl8r43ehVNZIU/HEI6OcFqwMG9pJV4I=

log/log.go

@@ -22,7 +22,10 @@ var LogSize int64
 var LogChannelCap int
 var LogPath string
 var LogLevel slog.Level = LevelTrace
+
+
 var gLogger, _ = NewTextLogger(LevelDebug, "", "",true,LogChannelCap)
+var isSetLogger bool
 var memPool = bytespool.NewMemAreaPool()
 
 // levels
@@ -37,6 +40,21 @@ const (
 	LevelFatal 				= slog.Level(20)
 )
 
+type ILogger interface {
+	Trace(msg string, args ...any)
+	Debug(msg string, args ...any)
+	Info(msg string, args ...any)
+	Warning(msg string, args ...any)
+	Error(msg string, args ...any)
+	Stack(msg string, args ...any)
+	Dump(msg string, args ...any)
+	Fatal(msg string, args ...any)
+
+	DoSPrintf(level slog.Level,a []interface{})
+	FormatHeader(buf *Buffer,level slog.Level,calldepth int)
+	Close()
+}
+
 type Logger struct {
 	Slogger *slog.Logger
 
@@ -47,7 +65,6 @@ type Logger struct {
 
 type IoWriter struct {
 	outFile    io.Writer  // destination for output
-	outConsole io.Writer  //os.Stdout
 	writeBytes  int64
 	logChannel chan []byte
 	wg             sync.WaitGroup
@@ -122,8 +139,8 @@ func (iw *IoWriter) Write(p []byte) (n int, err error){
 func (iw *IoWriter) writeIo(p []byte)  (n int, err error){
 	n,err = iw.writeFile(p)
 
-	if iw.outConsole != nil {
-		n,err = iw.outConsole.Write(p)
+	if OpenConsole {
+		n,err = os.Stdout.Write(p)
 	}
 
 	return
@@ -217,17 +234,12 @@ func (iw *IoWriter) swichFile() error{
 		iw.fileDay = now.Day()
 		iw.fileCreateTime = now.Unix()
 		atomic.StoreInt64(&iw.writeBytes,0)
-		if OpenConsole == true {
-			iw.outConsole = os.Stdout
-		}
-	}else{
-		iw.outConsole = os.Stdout
 	}
 
 	return nil
 }
 
-func NewTextLogger(level slog.Level,pathName string,filePrefix string,addSource bool,logChannelCap int) (*Logger,error){
+func NewTextLogger(level slog.Level,pathName string,filePrefix string,addSource bool,logChannelCap int) (ILogger,error){
 	var logger Logger
 	logger.ioWriter.filePath = pathName
 	logger.ioWriter.fileprefix = filePrefix
@@ -242,7 +254,7 @@ func NewTextLogger(level slog.Level,pathName string,filePrefix string,addSource
 	return &logger,nil
 }
 
-func NewJsonLogger(level slog.Level,pathName string,filePrefix string,addSource bool,logChannelCap int) (*Logger,error){
+func NewJsonLogger(level slog.Level,pathName string,filePrefix string,addSource bool,logChannelCap int) (ILogger,error){
 	var logger Logger
 	logger.ioWriter.filePath = pathName
 	logger.ioWriter.fileprefix = filePrefix
@@ -296,13 +308,18 @@ func (logger *Logger) Fatal(msg string, args ...any) {
 	os.Exit(1)
 }
 
-// It's dangerous to call the method on logging
-func Export(logger *Logger) {
-	if logger != nil {
+// It's non-thread-safe
+func SetLogger(logger ILogger) {
+	if logger != nil && isSetLogger == false {
 		gLogger = logger
+		isSetLogger = true
 	}
 }
 
+func GetLogger() ILogger{
+	return gLogger
+}
+
 func Trace(msg string, args ...any){
 	gLogger.Trace(msg, args...)
 }
@@ -415,7 +432,7 @@ func Group(key string, args ...any) slog.Attr {
 	return slog.Group(key, args...)
 }
 
-func (logger *Logger) doSPrintf(level slog.Level,a []interface{}) {
+func (logger *Logger) DoSPrintf(level slog.Level,a []interface{}) {
 	if logger.Slogger.Enabled(context.Background(),level) == false{
 		return
 	}
@@ -425,7 +442,7 @@ func (logger *Logger) doSPrintf(level slog.Level,a []interface{}) {
 
 	logger.sBuff.Reset()
 
-	logger.formatHeader(&logger.sBuff,level,3)
+	logger.FormatHeader(&logger.sBuff,level,3)
 
 	for _,s := range a {
 		logger.sBuff.AppendString(slog.AnyValue(s).String())
@@ -435,46 +452,46 @@ func (logger *Logger) doSPrintf(level slog.Level,a []interface{}) {
 }
 
  func (logger *Logger) STrace(a ...interface{}) {
-	 logger.doSPrintf(LevelTrace,a)
+	 logger.DoSPrintf(LevelTrace,a)
 }
 
 func (logger *Logger)  SDebug(a ...interface{}) {
-	logger.doSPrintf(LevelDebug,a)
+	logger.DoSPrintf(LevelDebug,a)
 }
 
 func (logger *Logger)  SInfo(a ...interface{}) {
-	logger.doSPrintf(LevelInfo,a)
+	logger.DoSPrintf(LevelInfo,a)
 }
 
 func (logger *Logger)  SWarning(a ...interface{}) {
-	logger.doSPrintf(LevelWarning,a)
+	logger.DoSPrintf(LevelWarning,a)
 }
 
 func (logger *Logger)  SError(a ...interface{}) {
-	logger.doSPrintf(LevelError,a)
+	logger.DoSPrintf(LevelError,a)
 }
 
 func STrace(a ...interface{}) {
-	gLogger.doSPrintf(LevelTrace,a)
+	gLogger.DoSPrintf(LevelTrace,a)
 }
 
 func SDebug(a ...interface{}) {
-	gLogger.doSPrintf(LevelDebug,a)
+	gLogger.DoSPrintf(LevelDebug,a)
 }
 
 func SInfo(a ...interface{}) {
-	gLogger.doSPrintf(LevelInfo,a)
+	gLogger.DoSPrintf(LevelInfo,a)
 }
 
 func SWarning(a ...interface{}) {
-	gLogger.doSPrintf(LevelWarning,a)
+	gLogger.DoSPrintf(LevelWarning,a)
 }
 
 func SError(a ...interface{}) {
-	gLogger.doSPrintf(LevelError,a)
+	gLogger.DoSPrintf(LevelError,a)
 }
 
-func (logger *Logger) formatHeader(buf *Buffer,level slog.Level,calldepth int) {
+func (logger *Logger) FormatHeader(buf *Buffer,level slog.Level,calldepth int) {
 	t := time.Now()
 	var file string
 	var line int

node/node.go

@@ -19,6 +19,7 @@ import (
 	"strings"
 	"syscall"
 	"time"
+	"github.com/duanhf2012/origin/util/sysprocess"
 )
 
 var sig chan os.Signal
@@ -28,6 +29,10 @@ var profilerInterval time.Duration
 var bValid bool
 var configDir = "./config/"
 
+const(
+	SingleStop   syscall.Signal = 10
+	SignalRetire syscall.Signal = 12
+)
 
 type BuildOSType = int8
 
@@ -38,13 +43,14 @@ const(
 )
 
 func init() {
-	sig = make(chan os.Signal, 3)
-	signal.Notify(sig, syscall.SIGINT, syscall.SIGTERM, syscall.Signal(10))
+	sig = make(chan os.Signal, 4)
+	signal.Notify(sig, syscall.SIGINT, syscall.SIGTERM, SingleStop,SignalRetire)
 
 	console.RegisterCommandBool("help", false, "<-help> This help.", usage)
 	console.RegisterCommandString("name", "", "<-name nodeName> Node's name.", setName)
 	console.RegisterCommandString("start", "", "<-start nodeid=nodeid> Run originserver.", startNode)
 	console.RegisterCommandString("stop", "", "<-stop nodeid=nodeid> Stop originserver process.", stopNode)
+	console.RegisterCommandString("retire", "", "<-retire nodeid=nodeid> retire originserver process.", retireNode)
 	console.RegisterCommandString("config", "", "<-config path> Configuration file path.", setConfigPath)
 	console.RegisterCommandString("console", "", "<-console true|false> Turn on or off screen log output.", openConsole)
 	console.RegisterCommandString("loglevel", "debug", "<-loglevel debug|release|warning|error|fatal> Set loglevel.", setLevel)
@@ -54,6 +60,11 @@ func init() {
 	console.RegisterCommandString("pprof", "", "<-pprof ip:port> Open performance analysis.", setPprof)
 }
 
+
+func notifyAllServiceRetire(){
+	service.NotifyAllServiceRetire()
+}
+
 func usage(val interface{}) error {
 	ret := val.(bool)
 	if ret == false {
@@ -189,7 +200,7 @@ func initLog() error {
 		fmt.Printf("cannot create log file!\n")
 		return err
 	}
-	log.Export(logger)
+	log.SetLogger(logger)
 	return nil
 }
 
@@ -201,6 +212,37 @@ func Start() {
 	}
 }
 
+
+func retireNode(args interface{}) error {
+	//1.解析参数
+	param := args.(string)
+	if param == "" {
+		return nil
+	}
+
+	sParam := strings.Split(param, "=")
+	if len(sParam) != 2 {
+		return fmt.Errorf("invalid option %s", param)
+	}
+	if sParam[0] != "nodeid" {
+		return fmt.Errorf("invalid option %s", param)
+	}
+	nId, err := strconv.Atoi(sParam[1])
+	if err != nil {
+		return fmt.Errorf("invalid option %s", param)
+	}
+
+	processId, err := getRunProcessPid(nId)
+	if err != nil {
+		return err
+	}
+
+
+	RetireProcess(processId)
+	return nil
+}
+
+
 func stopNode(args interface{}) error {
 	//1.解析参数
 	param := args.(string)
@@ -215,12 +257,12 @@ func stopNode(args interface{}) error {
 	if sParam[0] != "nodeid" {
 		return fmt.Errorf("invalid option %s", param)
 	}
-	nodeId, err := strconv.Atoi(sParam[1])
+	nId, err := strconv.Atoi(sParam[1])
 	if err != nil {
 		return fmt.Errorf("invalid option %s", param)
 	}
 
-	processId, err := getRunProcessPid(nodeId)
+	processId, err := getRunProcessPid(nId)
 	if err != nil {
 		return err
 	}
@@ -247,20 +289,43 @@ func startNode(args interface{}) error {
 	if err != nil {
 		return fmt.Errorf("invalid option %s", param)
 	}
+	for{
+		processId, pErr := getRunProcessPid(nodeId)
+		if pErr != nil {
+			break
+		}
 
-	timer.StartTimer(10*time.Millisecond, 1000000)
+		name, cErr := sysprocess.GetProcessNameByPID(int32(processId))
+		myName, mErr := sysprocess.GetMyProcessName()
+		//当前进程名获取失败，不应该发生
+		if mErr != nil {
+			log.SInfo("get my process's name is error,", err.Error())
+			os.Exit(-1)
+		}
+
+		//进程id存在，而且进程名也相同，被认为是当前进程重复运行
+		if cErr == nil && name == myName {
+			log.SInfo(fmt.Sprintf("repeat runs are not allowed,node is %d,processid is %d",nodeId,processId))
+			os.Exit(-1)
+		}
+		break
+	}
+
+	//2.记录进程id号
 	log.Info("Start running server.")
-	//2.初始化node
+	writeProcessPid(nodeId)
+	timer.StartTimer(10*time.Millisecond, 1000000)
+
+	//3.初始化node
 	initNode(nodeId)
 
-	//3.运行service
+	//4.运行service
 	service.Start()
 
-	//4.运行集群
+	//5.运行集群
 	cluster.GetCluster().Start()
 
-	//5.记录进程id号
-	writeProcessPid(nodeId)
+
 
 	//6.监听程序退出信号&性能报告
 	bRun := true
@@ -268,15 +333,23 @@ func startNode(args interface{}) error {
 	if profilerInterval > 0 {
 		pProfilerTicker = time.NewTicker(profilerInterval)
 	}
+
 	for bRun {
 		select {
-		case <-sig:
-			log.Info("receipt stop signal.")
-			bRun = false
+		case s := <-sig:
+			signal := s.(syscall.Signal)
+			if signal == SignalRetire {
+				log.Info("receipt retire signal.")
+				notifyAllServiceRetire()
+			}else {
+				bRun = false
+				log.Info("receipt stop signal.")
+			}
 		case <-pProfilerTicker.C:
 			profiler.Report()
 		}
 	}
+
 	cluster.GetCluster().Stop()
 	//7.退出
 	service.StopAllService()

node/node_linux.go

@@ -8,7 +8,7 @@ import (
 )
 
 func KillProcess(processId int){
-	err := syscall.Kill(processId,syscall.Signal(10))
+	err := syscall.Kill(processId,SingleStop)
 	if err != nil {
 		fmt.Printf("kill processid %d is fail:%+v.\n",processId,err)
 	}else{
@@ -19,3 +19,12 @@ func KillProcess(processId int){
 func GetBuildOSType() BuildOSType{
 	return Linux
 }
+
+func RetireProcess(processId int){
+	err := syscall.Kill(processId,SignalRetire)
+	if err != nil {
+		fmt.Printf("retire processid %d is fail:%+v.\n",processId,err)
+	}else{
+		fmt.Printf("retire processid %d is successful.\n",processId)
+	}
+}

node/node_mac.go

@@ -8,7 +8,7 @@ import (
 )
 
 func KillProcess(processId int){
-	err := syscall.Kill(processId,syscall.Signal(10))
+	err := syscall.Kill(processId,SingleStop)
 	if err != nil {
 		fmt.Printf("kill processid %d is fail:%+v.\n",processId,err)
 	}else{
@@ -19,3 +19,12 @@ func KillProcess(processId int){
 func GetBuildOSType() BuildOSType{
 	return Mac
 }
+
+func RetireProcess(processId int){
+	err := syscall.Kill(processId,SignalRetire)
+	if err != nil {
+		fmt.Printf("retire processid %d is fail:%+v.\n",processId,err)
+	}else{
+		fmt.Printf("retire processid %d is successful.\n",processId)
+	}
+}

node/node_win.go

@@ -2,10 +2,28 @@
 
 package node
 
-func KillProcess(processId int){
+import (
+	"os"
+	"fmt"
+)
 
+func KillProcess(processId int){
+	procss,err := os.FindProcess(processId)
+	if err != nil {
+		fmt.Printf("kill processid %d is fail:%+v.\n",processId,err)
+		return
+	}
+
+	err = procss.Kill()
+	if err != nil {
+		fmt.Printf("kill processid %d is fail:%+v.\n",processId,err)
+	}
 }
 
 func GetBuildOSType() BuildOSType{
 	return Windows
 }
+
+func RetireProcess(processId int){
+	fmt.Printf("This command does not support Windows")
+}

rpc/dynamicdiscover.pb.go

@@ -1,8 +1,8 @@
 // Code generated by protoc-gen-go. DO NOT EDIT.
 // versions:
 // 	protoc-gen-go v1.31.0
-// 	protoc        v3.11.4
-// source: test/rpc/dynamicdiscover.proto
+// 	protoc        v4.24.0
+// source: proto/rpcproto/dynamicdiscover.proto
 
 package rpc
 
@@ -30,13 +30,14 @@ type NodeInfo struct {
 	ListenAddr        string   `protobuf:"bytes,3,opt,name=ListenAddr,proto3" json:"ListenAddr,omitempty"`
 	MaxRpcParamLen    uint32   `protobuf:"varint,4,opt,name=MaxRpcParamLen,proto3" json:"MaxRpcParamLen,omitempty"`
 	Private           bool     `protobuf:"varint,5,opt,name=Private,proto3" json:"Private,omitempty"`
-	PublicServiceList []string `protobuf:"bytes,6,rep,name=PublicServiceList,proto3" json:"PublicServiceList,omitempty"`
+	Retire            bool     `protobuf:"varint,6,opt,name=Retire,proto3" json:"Retire,omitempty"`
+	PublicServiceList []string `protobuf:"bytes,7,rep,name=PublicServiceList,proto3" json:"PublicServiceList,omitempty"`
 }
 
 func (x *NodeInfo) Reset() {
 	*x = NodeInfo{}
 	if protoimpl.UnsafeEnabled {
-		mi := &file_test_rpc_dynamicdiscover_proto_msgTypes[0]
+		mi := &file_proto_rpcproto_dynamicdiscover_proto_msgTypes[0]
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		ms.StoreMessageInfo(mi)
 	}
@@ -49,7 +50,7 @@ func (x *NodeInfo) String() string {
 func (*NodeInfo) ProtoMessage() {}
 
 func (x *NodeInfo) ProtoReflect() protoreflect.Message {
-	mi := &file_test_rpc_dynamicdiscover_proto_msgTypes[0]
+	mi := &file_proto_rpcproto_dynamicdiscover_proto_msgTypes[0]
 	if protoimpl.UnsafeEnabled && x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -62,7 +63,7 @@ func (x *NodeInfo) ProtoReflect() protoreflect.Message {
 
 // Deprecated: Use NodeInfo.ProtoReflect.Descriptor instead.
 func (*NodeInfo) Descriptor() ([]byte, []int) {
-	return file_test_rpc_dynamicdiscover_proto_rawDescGZIP(), []int{0}
+	return file_proto_rpcproto_dynamicdiscover_proto_rawDescGZIP(), []int{0}
 }
 
 func (x *NodeInfo) GetNodeId() int32 {
@@ -100,6 +101,13 @@ func (x *NodeInfo) GetPrivate() bool {
 	return false
 }
 
+func (x *NodeInfo) GetRetire() bool {
+	if x != nil {
+		return x.Retire
+	}
+	return false
+}
+
 func (x *NodeInfo) GetPublicServiceList() []string {
 	if x != nil {
 		return x.PublicServiceList
@@ -119,7 +127,7 @@ type ServiceDiscoverReq struct {
 func (x *ServiceDiscoverReq) Reset() {
 	*x = ServiceDiscoverReq{}
 	if protoimpl.UnsafeEnabled {
-		mi := &file_test_rpc_dynamicdiscover_proto_msgTypes[1]
+		mi := &file_proto_rpcproto_dynamicdiscover_proto_msgTypes[1]
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		ms.StoreMessageInfo(mi)
 	}
@@ -132,7 +140,7 @@ func (x *ServiceDiscoverReq) String() string {
 func (*ServiceDiscoverReq) ProtoMessage() {}
 
 func (x *ServiceDiscoverReq) ProtoReflect() protoreflect.Message {
-	mi := &file_test_rpc_dynamicdiscover_proto_msgTypes[1]
+	mi := &file_proto_rpcproto_dynamicdiscover_proto_msgTypes[1]
 	if protoimpl.UnsafeEnabled && x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -145,7 +153,7 @@ func (x *ServiceDiscoverReq) ProtoReflect() protoreflect.Message {
 
 // Deprecated: Use ServiceDiscoverReq.ProtoReflect.Descriptor instead.
 func (*ServiceDiscoverReq) Descriptor() ([]byte, []int) {
-	return file_test_rpc_dynamicdiscover_proto_rawDescGZIP(), []int{1}
+	return file_proto_rpcproto_dynamicdiscover_proto_rawDescGZIP(), []int{1}
 }
 
 func (x *ServiceDiscoverReq) GetNodeInfo() *NodeInfo {
@@ -170,7 +178,7 @@ type SubscribeDiscoverNotify struct {
 func (x *SubscribeDiscoverNotify) Reset() {
 	*x = SubscribeDiscoverNotify{}
 	if protoimpl.UnsafeEnabled {
-		mi := &file_test_rpc_dynamicdiscover_proto_msgTypes[2]
+		mi := &file_proto_rpcproto_dynamicdiscover_proto_msgTypes[2]
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		ms.StoreMessageInfo(mi)
 	}
@@ -183,7 +191,7 @@ func (x *SubscribeDiscoverNotify) String() string {
 func (*SubscribeDiscoverNotify) ProtoMessage() {}
 
 func (x *SubscribeDiscoverNotify) ProtoReflect() protoreflect.Message {
-	mi := &file_test_rpc_dynamicdiscover_proto_msgTypes[2]
+	mi := &file_proto_rpcproto_dynamicdiscover_proto_msgTypes[2]
 	if protoimpl.UnsafeEnabled && x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -196,7 +204,7 @@ func (x *SubscribeDiscoverNotify) ProtoReflect() protoreflect.Message {
 
 // Deprecated: Use SubscribeDiscoverNotify.ProtoReflect.Descriptor instead.
 func (*SubscribeDiscoverNotify) Descriptor() ([]byte, []int) {
-	return file_test_rpc_dynamicdiscover_proto_rawDescGZIP(), []int{2}
+	return file_proto_rpcproto_dynamicdiscover_proto_rawDescGZIP(), []int{2}
 }
 
 func (x *SubscribeDiscoverNotify) GetMasterNodeId() int32 {
@@ -227,6 +235,54 @@ func (x *SubscribeDiscoverNotify) GetNodeInfo() []*NodeInfo {
 	return nil
 }
 
+// Client->Master
+type NodeRetireReq struct {
+	state         protoimpl.MessageState
+	sizeCache     protoimpl.SizeCache
+	unknownFields protoimpl.UnknownFields
+
+	NodeInfo *NodeInfo `protobuf:"bytes,1,opt,name=nodeInfo,proto3" json:"nodeInfo,omitempty"`
+}
+
+func (x *NodeRetireReq) Reset() {
+	*x = NodeRetireReq{}
+	if protoimpl.UnsafeEnabled {
+		mi := &file_proto_rpcproto_dynamicdiscover_proto_msgTypes[3]
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		ms.StoreMessageInfo(mi)
+	}
+}
+
+func (x *NodeRetireReq) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*NodeRetireReq) ProtoMessage() {}
+
+func (x *NodeRetireReq) ProtoReflect() protoreflect.Message {
+	mi := &file_proto_rpcproto_dynamicdiscover_proto_msgTypes[3]
+	if protoimpl.UnsafeEnabled && x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use NodeRetireReq.ProtoReflect.Descriptor instead.
+func (*NodeRetireReq) Descriptor() ([]byte, []int) {
+	return file_proto_rpcproto_dynamicdiscover_proto_rawDescGZIP(), []int{3}
+}
+
+func (x *NodeRetireReq) GetNodeInfo() *NodeInfo {
+	if x != nil {
+		return x.NodeInfo
+	}
+	return nil
+}
+
 // Master->Client
 type Empty struct {
 	state         protoimpl.MessageState
@@ -237,7 +293,7 @@ type Empty struct {
 func (x *Empty) Reset() {
 	*x = Empty{}
 	if protoimpl.UnsafeEnabled {
-		mi := &file_test_rpc_dynamicdiscover_proto_msgTypes[3]
+		mi := &file_proto_rpcproto_dynamicdiscover_proto_msgTypes[4]
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		ms.StoreMessageInfo(mi)
 	}
@@ -250,7 +306,7 @@ func (x *Empty) String() string {
 func (*Empty) ProtoMessage() {}
 
 func (x *Empty) ProtoReflect() protoreflect.Message {
-	mi := &file_test_rpc_dynamicdiscover_proto_msgTypes[3]
+	mi := &file_proto_rpcproto_dynamicdiscover_proto_msgTypes[4]
 	if protoimpl.UnsafeEnabled && x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -263,82 +319,89 @@ func (x *Empty) ProtoReflect() protoreflect.Message {
 
 // Deprecated: Use Empty.ProtoReflect.Descriptor instead.
 func (*Empty) Descriptor() ([]byte, []int) {
-	return file_test_rpc_dynamicdiscover_proto_rawDescGZIP(), []int{3}
+	return file_proto_rpcproto_dynamicdiscover_proto_rawDescGZIP(), []int{4}
 }
 
-var File_test_rpc_dynamicdiscover_proto protoreflect.FileDescriptor
+var File_proto_rpcproto_dynamicdiscover_proto protoreflect.FileDescriptor
 
-var file_test_rpc_dynamicdiscover_proto_rawDesc = []byte{
-	0x0a, 0x1e, 0x74, 0x65, 0x73, 0x74, 0x2f, 0x72, 0x70, 0x63, 0x2f, 0x64, 0x79, 0x6e, 0x61, 0x6d,
-	0x69, 0x63, 0x64, 0x69, 0x73, 0x63, 0x6f, 0x76, 0x65, 0x72, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f,
-	0x12, 0x03, 0x72, 0x70, 0x63, 0x22, 0xce, 0x01, 0x0a, 0x08, 0x4e, 0x6f, 0x64, 0x65, 0x49, 0x6e,
-	0x66, 0x6f, 0x12, 0x16, 0x0a, 0x06, 0x4e, 0x6f, 0x64, 0x65, 0x49, 0x64, 0x18, 0x01, 0x20, 0x01,
-	0x28, 0x05, 0x52, 0x06, 0x4e, 0x6f, 0x64, 0x65, 0x49, 0x64, 0x12, 0x1a, 0x0a, 0x08, 0x4e, 0x6f,
-	0x64, 0x65, 0x4e, 0x61, 0x6d, 0x65, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x08, 0x4e, 0x6f,
-	0x64, 0x65, 0x4e, 0x61, 0x6d, 0x65, 0x12, 0x1e, 0x0a, 0x0a, 0x4c, 0x69, 0x73, 0x74, 0x65, 0x6e,
-	0x41, 0x64, 0x64, 0x72, 0x18, 0x03, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0a, 0x4c, 0x69, 0x73, 0x74,
-	0x65, 0x6e, 0x41, 0x64, 0x64, 0x72, 0x12, 0x26, 0x0a, 0x0e, 0x4d, 0x61, 0x78, 0x52, 0x70, 0x63,
-	0x50, 0x61, 0x72, 0x61, 0x6d, 0x4c, 0x65, 0x6e, 0x18, 0x04, 0x20, 0x01, 0x28, 0x0d, 0x52, 0x0e,
-	0x4d, 0x61, 0x78, 0x52, 0x70, 0x63, 0x50, 0x61, 0x72, 0x61, 0x6d, 0x4c, 0x65, 0x6e, 0x12, 0x18,
-	0x0a, 0x07, 0x50, 0x72, 0x69, 0x76, 0x61, 0x74, 0x65, 0x18, 0x05, 0x20, 0x01, 0x28, 0x08, 0x52,
-	0x07, 0x50, 0x72, 0x69, 0x76, 0x61, 0x74, 0x65, 0x12, 0x2c, 0x0a, 0x11, 0x50, 0x75, 0x62, 0x6c,
-	0x69, 0x63, 0x53, 0x65, 0x72, 0x76, 0x69, 0x63, 0x65, 0x4c, 0x69, 0x73, 0x74, 0x18, 0x06, 0x20,
-	0x03, 0x28, 0x09, 0x52, 0x11, 0x50, 0x75, 0x62, 0x6c, 0x69, 0x63, 0x53, 0x65, 0x72, 0x76, 0x69,
-	0x63, 0x65, 0x4c, 0x69, 0x73, 0x74, 0x22, 0x3f, 0x0a, 0x12, 0x53, 0x65, 0x72, 0x76, 0x69, 0x63,
-	0x65, 0x44, 0x69, 0x73, 0x63, 0x6f, 0x76, 0x65, 0x72, 0x52, 0x65, 0x71, 0x12, 0x29, 0x0a, 0x08,
-	0x6e, 0x6f, 0x64, 0x65, 0x49, 0x6e, 0x66, 0x6f, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x0d,
-	0x2e, 0x72, 0x70, 0x63, 0x2e, 0x4e, 0x6f, 0x64, 0x65, 0x49, 0x6e, 0x66, 0x6f, 0x52, 0x08, 0x6e,
-	0x6f, 0x64, 0x65, 0x49, 0x6e, 0x66, 0x6f, 0x22, 0x9e, 0x01, 0x0a, 0x17, 0x53, 0x75, 0x62, 0x73,
-	0x63, 0x72, 0x69, 0x62, 0x65, 0x44, 0x69, 0x73, 0x63, 0x6f, 0x76, 0x65, 0x72, 0x4e, 0x6f, 0x74,
-	0x69, 0x66, 0x79, 0x12, 0x22, 0x0a, 0x0c, 0x4d, 0x61, 0x73, 0x74, 0x65, 0x72, 0x4e, 0x6f, 0x64,
-	0x65, 0x49, 0x64, 0x18, 0x01, 0x20, 0x01, 0x28, 0x05, 0x52, 0x0c, 0x4d, 0x61, 0x73, 0x74, 0x65,
-	0x72, 0x4e, 0x6f, 0x64, 0x65, 0x49, 0x64, 0x12, 0x16, 0x0a, 0x06, 0x49, 0x73, 0x46, 0x75, 0x6c,
-	0x6c, 0x18, 0x02, 0x20, 0x01, 0x28, 0x08, 0x52, 0x06, 0x49, 0x73, 0x46, 0x75, 0x6c, 0x6c, 0x12,
-	0x1c, 0x0a, 0x09, 0x44, 0x65, 0x6c, 0x4e, 0x6f, 0x64, 0x65, 0x49, 0x64, 0x18, 0x03, 0x20, 0x01,
-	0x28, 0x05, 0x52, 0x09, 0x44, 0x65, 0x6c, 0x4e, 0x6f, 0x64, 0x65, 0x49, 0x64, 0x12, 0x29, 0x0a,
-	0x08, 0x6e, 0x6f, 0x64, 0x65, 0x49, 0x6e, 0x66, 0x6f, 0x18, 0x04, 0x20, 0x03, 0x28, 0x0b, 0x32,
-	0x0d, 0x2e, 0x72, 0x70, 0x63, 0x2e, 0x4e, 0x6f, 0x64, 0x65, 0x49, 0x6e, 0x66, 0x6f, 0x52, 0x08,
-	0x6e, 0x6f, 0x64, 0x65, 0x49, 0x6e, 0x66, 0x6f, 0x22, 0x07, 0x0a, 0x05, 0x45, 0x6d, 0x70, 0x74,
-	0x79, 0x42, 0x07, 0x5a, 0x05, 0x2e, 0x3b, 0x72, 0x70, 0x63, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74,
-	0x6f, 0x33,
+var file_proto_rpcproto_dynamicdiscover_proto_rawDesc = []byte{
+	0x0a, 0x24, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x2f, 0x72, 0x70, 0x63, 0x70, 0x72, 0x6f, 0x74, 0x6f,
+	0x2f, 0x64, 0x79, 0x6e, 0x61, 0x6d, 0x69, 0x63, 0x64, 0x69, 0x73, 0x63, 0x6f, 0x76, 0x65, 0x72,
+	0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x12, 0x03, 0x72, 0x70, 0x63, 0x22, 0xe6, 0x01, 0x0a, 0x08,
+	0x4e, 0x6f, 0x64, 0x65, 0x49, 0x6e, 0x66, 0x6f, 0x12, 0x16, 0x0a, 0x06, 0x4e, 0x6f, 0x64, 0x65,
+	0x49, 0x64, 0x18, 0x01, 0x20, 0x01, 0x28, 0x05, 0x52, 0x06, 0x4e, 0x6f, 0x64, 0x65, 0x49, 0x64,
+	0x12, 0x1a, 0x0a, 0x08, 0x4e, 0x6f, 0x64, 0x65, 0x4e, 0x61, 0x6d, 0x65, 0x18, 0x02, 0x20, 0x01,
+	0x28, 0x09, 0x52, 0x08, 0x4e, 0x6f, 0x64, 0x65, 0x4e, 0x61, 0x6d, 0x65, 0x12, 0x1e, 0x0a, 0x0a,
+	0x4c, 0x69, 0x73, 0x74, 0x65, 0x6e, 0x41, 0x64, 0x64, 0x72, 0x18, 0x03, 0x20, 0x01, 0x28, 0x09,
+	0x52, 0x0a, 0x4c, 0x69, 0x73, 0x74, 0x65, 0x6e, 0x41, 0x64, 0x64, 0x72, 0x12, 0x26, 0x0a, 0x0e,
+	0x4d, 0x61, 0x78, 0x52, 0x70, 0x63, 0x50, 0x61, 0x72, 0x61, 0x6d, 0x4c, 0x65, 0x6e, 0x18, 0x04,
+	0x20, 0x01, 0x28, 0x0d, 0x52, 0x0e, 0x4d, 0x61, 0x78, 0x52, 0x70, 0x63, 0x50, 0x61, 0x72, 0x61,
+	0x6d, 0x4c, 0x65, 0x6e, 0x12, 0x18, 0x0a, 0x07, 0x50, 0x72, 0x69, 0x76, 0x61, 0x74, 0x65, 0x18,
+	0x05, 0x20, 0x01, 0x28, 0x08, 0x52, 0x07, 0x50, 0x72, 0x69, 0x76, 0x61, 0x74, 0x65, 0x12, 0x16,
+	0x0a, 0x06, 0x52, 0x65, 0x74, 0x69, 0x72, 0x65, 0x18, 0x06, 0x20, 0x01, 0x28, 0x08, 0x52, 0x06,
+	0x52, 0x65, 0x74, 0x69, 0x72, 0x65, 0x12, 0x2c, 0x0a, 0x11, 0x50, 0x75, 0x62, 0x6c, 0x69, 0x63,
+	0x53, 0x65, 0x72, 0x76, 0x69, 0x63, 0x65, 0x4c, 0x69, 0x73, 0x74, 0x18, 0x07, 0x20, 0x03, 0x28,
+	0x09, 0x52, 0x11, 0x50, 0x75, 0x62, 0x6c, 0x69, 0x63, 0x53, 0x65, 0x72, 0x76, 0x69, 0x63, 0x65,
+	0x4c, 0x69, 0x73, 0x74, 0x22, 0x3f, 0x0a, 0x12, 0x53, 0x65, 0x72, 0x76, 0x69, 0x63, 0x65, 0x44,
+	0x69, 0x73, 0x63, 0x6f, 0x76, 0x65, 0x72, 0x52, 0x65, 0x71, 0x12, 0x29, 0x0a, 0x08, 0x6e, 0x6f,
+	0x64, 0x65, 0x49, 0x6e, 0x66, 0x6f, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x0d, 0x2e, 0x72,
+	0x70, 0x63, 0x2e, 0x4e, 0x6f, 0x64, 0x65, 0x49, 0x6e, 0x66, 0x6f, 0x52, 0x08, 0x6e, 0x6f, 0x64,
+	0x65, 0x49, 0x6e, 0x66, 0x6f, 0x22, 0x9e, 0x01, 0x0a, 0x17, 0x53, 0x75, 0x62, 0x73, 0x63, 0x72,
+	0x69, 0x62, 0x65, 0x44, 0x69, 0x73, 0x63, 0x6f, 0x76, 0x65, 0x72, 0x4e, 0x6f, 0x74, 0x69, 0x66,
+	0x79, 0x12, 0x22, 0x0a, 0x0c, 0x4d, 0x61, 0x73, 0x74, 0x65, 0x72, 0x4e, 0x6f, 0x64, 0x65, 0x49,
+	0x64, 0x18, 0x01, 0x20, 0x01, 0x28, 0x05, 0x52, 0x0c, 0x4d, 0x61, 0x73, 0x74, 0x65, 0x72, 0x4e,
+	0x6f, 0x64, 0x65, 0x49, 0x64, 0x12, 0x16, 0x0a, 0x06, 0x49, 0x73, 0x46, 0x75, 0x6c, 0x6c, 0x18,
+	0x02, 0x20, 0x01, 0x28, 0x08, 0x52, 0x06, 0x49, 0x73, 0x46, 0x75, 0x6c, 0x6c, 0x12, 0x1c, 0x0a,
+	0x09, 0x44, 0x65, 0x6c, 0x4e, 0x6f, 0x64, 0x65, 0x49, 0x64, 0x18, 0x03, 0x20, 0x01, 0x28, 0x05,
+	0x52, 0x09, 0x44, 0x65, 0x6c, 0x4e, 0x6f, 0x64, 0x65, 0x49, 0x64, 0x12, 0x29, 0x0a, 0x08, 0x6e,
+	0x6f, 0x64, 0x65, 0x49, 0x6e, 0x66, 0x6f, 0x18, 0x04, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x0d, 0x2e,
+	0x72, 0x70, 0x63, 0x2e, 0x4e, 0x6f, 0x64, 0x65, 0x49, 0x6e, 0x66, 0x6f, 0x52, 0x08, 0x6e, 0x6f,
+	0x64, 0x65, 0x49, 0x6e, 0x66, 0x6f, 0x22, 0x3a, 0x0a, 0x0d, 0x4e, 0x6f, 0x64, 0x65, 0x52, 0x65,
+	0x74, 0x69, 0x72, 0x65, 0x52, 0x65, 0x71, 0x12, 0x29, 0x0a, 0x08, 0x6e, 0x6f, 0x64, 0x65, 0x49,
+	0x6e, 0x66, 0x6f, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x0d, 0x2e, 0x72, 0x70, 0x63, 0x2e,
+	0x4e, 0x6f, 0x64, 0x65, 0x49, 0x6e, 0x66, 0x6f, 0x52, 0x08, 0x6e, 0x6f, 0x64, 0x65, 0x49, 0x6e,
+	0x66, 0x6f, 0x22, 0x07, 0x0a, 0x05, 0x45, 0x6d, 0x70, 0x74, 0x79, 0x42, 0x07, 0x5a, 0x05, 0x2e,
+	0x3b, 0x72, 0x70, 0x63, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
 }
 
 var (
-	file_test_rpc_dynamicdiscover_proto_rawDescOnce sync.Once
-	file_test_rpc_dynamicdiscover_proto_rawDescData = file_test_rpc_dynamicdiscover_proto_rawDesc
+	file_proto_rpcproto_dynamicdiscover_proto_rawDescOnce sync.Once
+	file_proto_rpcproto_dynamicdiscover_proto_rawDescData = file_proto_rpcproto_dynamicdiscover_proto_rawDesc
 )
 
-func file_test_rpc_dynamicdiscover_proto_rawDescGZIP() []byte {
-	file_test_rpc_dynamicdiscover_proto_rawDescOnce.Do(func() {
-		file_test_rpc_dynamicdiscover_proto_rawDescData = protoimpl.X.CompressGZIP(file_test_rpc_dynamicdiscover_proto_rawDescData)
+func file_proto_rpcproto_dynamicdiscover_proto_rawDescGZIP() []byte {
+	file_proto_rpcproto_dynamicdiscover_proto_rawDescOnce.Do(func() {
+		file_proto_rpcproto_dynamicdiscover_proto_rawDescData = protoimpl.X.CompressGZIP(file_proto_rpcproto_dynamicdiscover_proto_rawDescData)
 	})
-	return file_test_rpc_dynamicdiscover_proto_rawDescData
+	return file_proto_rpcproto_dynamicdiscover_proto_rawDescData
 }
 
-var file_test_rpc_dynamicdiscover_proto_msgTypes = make([]protoimpl.MessageInfo, 4)
-var file_test_rpc_dynamicdiscover_proto_goTypes = []interface{}{
+var file_proto_rpcproto_dynamicdiscover_proto_msgTypes = make([]protoimpl.MessageInfo, 5)
+var file_proto_rpcproto_dynamicdiscover_proto_goTypes = []interface{}{
 	(*NodeInfo)(nil),                // 0: rpc.NodeInfo
 	(*ServiceDiscoverReq)(nil),      // 1: rpc.ServiceDiscoverReq
 	(*SubscribeDiscoverNotify)(nil), // 2: rpc.SubscribeDiscoverNotify
-	(*Empty)(nil),                   // 3: rpc.Empty
+	(*NodeRetireReq)(nil),           // 3: rpc.NodeRetireReq
+	(*Empty)(nil),                   // 4: rpc.Empty
 }
-var file_test_rpc_dynamicdiscover_proto_depIdxs = []int32{
+var file_proto_rpcproto_dynamicdiscover_proto_depIdxs = []int32{
 	0, // 0: rpc.ServiceDiscoverReq.nodeInfo:type_name -> rpc.NodeInfo
 	0, // 1: rpc.SubscribeDiscoverNotify.nodeInfo:type_name -> rpc.NodeInfo
-	2, // [2:2] is the sub-list for method output_type
-	2, // [2:2] is the sub-list for method input_type
-	2, // [2:2] is the sub-list for extension type_name
-	2, // [2:2] is the sub-list for extension extendee
-	0, // [0:2] is the sub-list for field type_name
+	0, // 2: rpc.NodeRetireReq.nodeInfo:type_name -> rpc.NodeInfo
+	3, // [3:3] is the sub-list for method output_type
+	3, // [3:3] is the sub-list for method input_type
+	3, // [3:3] is the sub-list for extension type_name
+	3, // [3:3] is the sub-list for extension extendee
+	0, // [0:3] is the sub-list for field type_name
 }
 
-func init() { file_test_rpc_dynamicdiscover_proto_init() }
-func file_test_rpc_dynamicdiscover_proto_init() {
-	if File_test_rpc_dynamicdiscover_proto != nil {
+func init() { file_proto_rpcproto_dynamicdiscover_proto_init() }
+func file_proto_rpcproto_dynamicdiscover_proto_init() {
+	if File_proto_rpcproto_dynamicdiscover_proto != nil {
 		return
 	}
 	if !protoimpl.UnsafeEnabled {
-		file_test_rpc_dynamicdiscover_proto_msgTypes[0].Exporter = func(v interface{}, i int) interface{} {
+		file_proto_rpcproto_dynamicdiscover_proto_msgTypes[0].Exporter = func(v interface{}, i int) interface{} {
 			switch v := v.(*NodeInfo); i {
 			case 0:
 				return &v.state
@@ -350,7 +413,7 @@ func file_test_rpc_dynamicdiscover_proto_init() {
 				return nil
 			}
 		}
-		file_test_rpc_dynamicdiscover_proto_msgTypes[1].Exporter = func(v interface{}, i int) interface{} {
+		file_proto_rpcproto_dynamicdiscover_proto_msgTypes[1].Exporter = func(v interface{}, i int) interface{} {
 			switch v := v.(*ServiceDiscoverReq); i {
 			case 0:
 				return &v.state
@@ -362,7 +425,7 @@ func file_test_rpc_dynamicdiscover_proto_init() {
 				return nil
 			}
 		}
-		file_test_rpc_dynamicdiscover_proto_msgTypes[2].Exporter = func(v interface{}, i int) interface{} {
+		file_proto_rpcproto_dynamicdiscover_proto_msgTypes[2].Exporter = func(v interface{}, i int) interface{} {
 			switch v := v.(*SubscribeDiscoverNotify); i {
 			case 0:
 				return &v.state
@@ -374,7 +437,19 @@ func file_test_rpc_dynamicdiscover_proto_init() {
 				return nil
 			}
 		}
-		file_test_rpc_dynamicdiscover_proto_msgTypes[3].Exporter = func(v interface{}, i int) interface{} {
+		file_proto_rpcproto_dynamicdiscover_proto_msgTypes[3].Exporter = func(v interface{}, i int) interface{} {
+			switch v := v.(*NodeRetireReq); i {
+			case 0:
+				return &v.state
+			case 1:
+				return &v.sizeCache
+			case 2:
+				return &v.unknownFields
+			default:
+				return nil
+			}
+		}
+		file_proto_rpcproto_dynamicdiscover_proto_msgTypes[4].Exporter = func(v interface{}, i int) interface{} {
 			switch v := v.(*Empty); i {
 			case 0:
 				return &v.state
@@ -391,18 +466,18 @@ func file_test_rpc_dynamicdiscover_proto_init() {
 	out := protoimpl.TypeBuilder{
 		File: protoimpl.DescBuilder{
 			GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
-			RawDescriptor: file_test_rpc_dynamicdiscover_proto_rawDesc,
+			RawDescriptor: file_proto_rpcproto_dynamicdiscover_proto_rawDesc,
 			NumEnums:      0,
-			NumMessages:   4,
+			NumMessages:   5,
 			NumExtensions: 0,
 			NumServices:   0,
 		},
-		GoTypes:           file_test_rpc_dynamicdiscover_proto_goTypes,
-		DependencyIndexes: file_test_rpc_dynamicdiscover_proto_depIdxs,
-		MessageInfos:      file_test_rpc_dynamicdiscover_proto_msgTypes,
+		GoTypes:           file_proto_rpcproto_dynamicdiscover_proto_goTypes,
+		DependencyIndexes: file_proto_rpcproto_dynamicdiscover_proto_depIdxs,
+		MessageInfos:      file_proto_rpcproto_dynamicdiscover_proto_msgTypes,
 	}.Build()
-	File_test_rpc_dynamicdiscover_proto = out.File
-	file_test_rpc_dynamicdiscover_proto_rawDesc = nil
-	file_test_rpc_dynamicdiscover_proto_goTypes = nil
-	file_test_rpc_dynamicdiscover_proto_depIdxs = nil
+	File_proto_rpcproto_dynamicdiscover_proto = out.File
+	file_proto_rpcproto_dynamicdiscover_proto_rawDesc = nil
+	file_proto_rpcproto_dynamicdiscover_proto_goTypes = nil
+	file_proto_rpcproto_dynamicdiscover_proto_depIdxs = nil
 }

rpc/dynamicdiscover.proto

@@ -8,7 +8,8 @@ message NodeInfo{
     string ListenAddr  = 3;
     uint32 MaxRpcParamLen = 4;
     bool Private = 5;
-    repeated string PublicServiceList = 6;
+	bool Retire = 6;
+    repeated string PublicServiceList = 7;
 }
 
 //Client->Master
@@ -24,6 +25,12 @@ message SubscribeDiscoverNotify{
     repeated NodeInfo nodeInfo = 4;
 }
 
+
+//Client->Master
+message NodeRetireReq{
+    NodeInfo nodeInfo = 1;
+}
+
 //Master->Client
 message Empty{
 }

rpc/pbprocessor.go

@@ -74,7 +74,7 @@ func (slf *PBProcessor) IsParse(param interface{}) bool {
 }
 
 func (slf *PBProcessor)	GetProcessorType() RpcProcessorType{
-	return RpcProcessorGoGoPB
+	return RpcProcessorPB
 }
 
 func (slf *PBProcessor) Clone(src interface{}) (interface{},error){

rpc/rpchandler.go

@@ -14,7 +14,7 @@ import (
 
 const maxClusterNode int = 128
 
-type FuncRpcClient func(nodeId int, serviceMethod string, client []*Client) (error, int)
+type FuncRpcClient func(nodeId int, serviceMethod string,filterRetire bool, client []*Client) (error, int)
 type FuncRpcServer func() *Server
 
 
@@ -73,7 +73,7 @@ type INodeListener interface {
 
 type IDiscoveryServiceListener interface {
 	OnDiscoveryService(nodeId int, serviceName []string)
-	OnUnDiscoveryService(nodeId int, serviceName []string)
+	OnUnDiscoveryService(nodeId int)
 }
 
 type CancelRpc func()
@@ -428,7 +428,7 @@ func (handler *RpcHandler) CallMethod(client *Client,ServiceMethod string, param
 
 func (handler *RpcHandler) goRpc(processor IRpcProcessor, bCast bool, nodeId int, serviceMethod string, args interface{}) error {
 	var pClientList [maxClusterNode]*Client
-	err, count := handler.funcRpcClient(nodeId, serviceMethod, pClientList[:])
+	err, count := handler.funcRpcClient(nodeId, serviceMethod,false, pClientList[:])
 	if count == 0 {
 		if err != nil {
 			log.Error("call serviceMethod is failed",log.String("serviceMethod",serviceMethod),log.ErrorAttr("error",err))
@@ -458,7 +458,7 @@ func (handler *RpcHandler) goRpc(processor IRpcProcessor, bCast bool, nodeId int
 
 func (handler *RpcHandler) callRpc(timeout time.Duration,nodeId int, serviceMethod string, args interface{}, reply interface{}) error {
 	var pClientList [maxClusterNode]*Client
-	err, count := handler.funcRpcClient(nodeId, serviceMethod, pClientList[:])
+	err, count := handler.funcRpcClient(nodeId, serviceMethod,false, pClientList[:])
 	if err != nil {
 		log.Error("Call serviceMethod is failed",log.ErrorAttr("error",err))
 		return err
@@ -502,7 +502,7 @@ func (handler *RpcHandler) asyncCallRpc(timeout time.Duration,nodeId int, servic
 
 	reply := reflect.New(fVal.Type().In(0).Elem()).Interface()
 	var pClientList [2]*Client
-	err, count := handler.funcRpcClient(nodeId, serviceMethod, pClientList[:])
+	err, count := handler.funcRpcClient(nodeId, serviceMethod,false, pClientList[:])
 	if count == 0 || err != nil {
 		if err == nil {
 			if nodeId  > 0 {
@@ -585,7 +585,7 @@ func (handler *RpcHandler) CastGo(serviceMethod string, args interface{}) error
 
 func (handler *RpcHandler) RawGoNode(rpcProcessorType RpcProcessorType, nodeId int, rpcMethodId uint32, serviceName string, rawArgs []byte) error {
 	processor := GetProcessor(uint8(rpcProcessorType))
-	err, count := handler.funcRpcClient(nodeId, serviceName, handler.pClientList)
+	err, count := handler.funcRpcClient(nodeId, serviceName,false, handler.pClientList)
 	if count == 0 || err != nil {
 		log.Error("call serviceMethod is failed",log.ErrorAttr("error",err))
 		return err

rpc/server.go

@@ -17,7 +17,7 @@ type RpcProcessorType uint8
 
 const (
 	RpcProcessorJson   RpcProcessorType = 0
-	RpcProcessorGoGoPB RpcProcessorType = 1
+	RpcProcessorPB RpcProcessorType = 1
 )
 
 var arrayProcessor = []IRpcProcessor{&JsonProcessor{}, &PBProcessor{}}

service/service.go

@@ -28,6 +28,7 @@ type IService interface {
 	OnSetup(iService IService)
 	OnInit() error
 	OnStart()
+	OnRetire()
 	OnRelease()
 
 	SetName(serviceName string)
@@ -40,6 +41,9 @@ type IService interface {
 
 	SetEventChannelNum(num int)
 	OpenProfiler()
+
+	SetRetire()     //设置服务退休状态
+	IsRetire() bool //服务是否退休
 }
 
 type Service struct {
@@ -51,6 +55,7 @@ type Service struct {
 	serviceCfg     interface{}
 	goroutineNum   int32
 	startStatus    bool
+	retire int32
 	eventProcessor event.IEventProcessor
 	profiler *profiler.Profiler //性能分析器
 	nodeEventLister rpc.INodeListener
@@ -97,6 +102,19 @@ func (s *Service) OpenProfiler()  {
 	}
 }
 
+func (s *Service) IsRetire() bool{
+	return atomic.LoadInt32(&s.retire) != 0
+}
+
+func (s *Service) SetRetire(){
+	atomic.StoreInt32(&s.retire,1)
+
+	ev := event.NewEvent()
+	ev.Type = event.Sys_Event_Retire
+
+	s.pushEvent(ev)
+}
+
 func (s *Service) Init(iService IService,getClientFun rpc.FuncRpcClient,getServerFun rpc.FuncRpcServer,serviceCfg interface{}) {
 	s.closeSig = make(chan struct{})
 	s.dispatcher =timer.NewDispatcher(timerDispatcherLen)
@@ -155,6 +173,9 @@ func (s *Service) Run() {
 			concurrent.DoCallback(cb)
 		case ev := <- s.chanEvent:
 			switch ev.GetEventType() {
+			case event.Sys_Event_Retire:
+				log.Info("service OnRetire",log.String("servceName",s.GetName()))
+				s.self.(IService).OnRetire()
 			case event.ServiceRpcRequestEvent:
 				cEvent,ok := ev.(*event.Event)
 				if ok == false {
@@ -304,7 +325,7 @@ func (s *Service) OnDiscoverServiceEvent(ev event.IEvent){
 	if event.IsDiscovery {
 		s.discoveryServiceLister.OnDiscoveryService(event.NodeId,event.ServiceName)
 	}else{
-		s.discoveryServiceLister.OnUnDiscoveryService(event.NodeId,event.ServiceName)
+		s.discoveryServiceLister.OnUnDiscoveryService(event.NodeId)
 	}
 }
 
@@ -387,3 +408,6 @@ func (s *Service) SetGoRoutineNum(goroutineNum int32) bool {
 	s.goroutineNum = goroutineNum
 	return true
 }
+
+func (s *Service) OnRetire(){
+}

service/servicemgr.go

@@ -60,3 +60,9 @@ func StopAllService(){
 		setupServiceList[i].Stop()
 	}
 }
+
+func NotifyAllServiceRetire(){
+	for i := len(setupServiceList) - 1; i >= 0; i-- {
+		setupServiceList[i].SetRetire()
+	}
+}

sysmodule/httpclientmodule/httpclientpoolmodule.go

@@ -43,7 +43,15 @@ func (slf *SyncHttpResponse) Get(timeoutMs int) HttpResponse {
 	}
 }
 
-func (m *HttpClientModule) Init(maxpool int, proxyUrl string) {
+func (m *HttpClientModule) InitHttpClient(transport http.RoundTripper,timeout time.Duration,checkRedirect func(req *http.Request, via []*http.Request) error){
+	m.client = &http.Client{
+		Transport: transport,
+		Timeout: timeout,
+		CheckRedirect: checkRedirect,
+	}
+}
+
+func (m *HttpClientModule) Init(proxyUrl string, maxpool int, dialTimeout time.Duration,dialKeepAlive time.Duration,idleConnTimeout time.Duration,timeout time.Duration) {
 	type ProxyFun func(_ *http.Request) (*url.URL, error)
 	var proxyFun ProxyFun
 	if proxyUrl != "" {
@@ -55,16 +63,16 @@ func (m *HttpClientModule) Init(maxpool int, proxyUrl string) {
 	m.client = &http.Client{
 		Transport: &http.Transport{
 			DialContext: (&net.Dialer{
-				Timeout:   5 * time.Second,
-				KeepAlive: 30 * time.Second,
+				Timeout:   dialTimeout,
+				KeepAlive: dialKeepAlive,
 			}).DialContext,
 			MaxIdleConns:        maxpool,
 			MaxIdleConnsPerHost: maxpool,
-			IdleConnTimeout:     60 * time.Second,
+			IdleConnTimeout:     idleConnTimeout,
 			Proxy:               proxyFun,
 			TLSClientConfig:     &tls.Config{InsecureSkipVerify: true},
 		},
-		Timeout: 5 * time.Second,
+		Timeout: timeout,
 	}
 }
 

sysservice/rankservice/MongodbPersist.go

@@ -263,7 +263,7 @@ func (mp *MongoPersist) JugeTimeoutSave() bool{
 
 func (mp *MongoPersist)  persistCoroutine(){
 	defer mp.waitGroup.Done()
-	for atomic.LoadInt32(&mp.stop)==0 || mp.hasPersistData(){
+	for atomic.LoadInt32(&mp.stop)==0 {
 		//间隔时间sleep
 		time.Sleep(time.Second*1)
 
@@ -357,7 +357,7 @@ func (mp *MongoPersist) removeRankData(rankId uint64,keys []uint64) bool {
 
 func (mp *MongoPersist) upsertToDB(collectName string,rankData *RankData) error{
 	condition := bson.D{{"_id", rankData.Key}}
-	upsert := bson.M{"_id":rankData.Key,"RefreshTime": rankData.refreshTimestamp, "SortData": rankData.SortData, "Data": rankData.Data,"ExData":rankData.ExData}
+	upsert := bson.M{"_id":rankData.Key,"RefreshTime": rankData.RefreshTimestamp, "SortData": rankData.SortData, "Data": rankData.Data,"ExData":rankData.ExData}
 	update := bson.M{"$set": upsert}
 
 	s := mp.mongo.TakeSession()

sysservice/rankservice/RankData.go

@@ -19,7 +19,7 @@ type RankData struct {
 	Data                 []byte
 	ExData               []int64
 
-	refreshTimestamp int64 //刷新时间
+	RefreshTimestamp int64 //刷新时间
 	//bRelease bool
 	ref         bool
 	compareFunc func(other skip.Comparator) int
@@ -39,7 +39,7 @@ func NewRankData(isDec bool, data *rpc.RankData,refreshTimestamp int64) *RankDat
 		ret.ExData = append(ret.ExData,d.InitValue+d.IncreaseValue)
 	}
 
-	ret.refreshTimestamp = refreshTimestamp
+	ret.RefreshTimestamp = refreshTimestamp
 
 	return ret
 }

sysservice/rankservice/RankSkip.go

@@ -250,7 +250,7 @@ func (rs *RankSkip) UpsetRank(upsetData *rpc.RankData, refreshTimestamp int64, f
 		//找到的情况对比排名数据是否有变化,无变化进行data更新,有变化则进行删除更新
 		if compareIsEqual(rankNode.SortData, upsetData.SortData) {
 			rankNode.Data = upsetData.GetData()
-			rankNode.refreshTimestamp = refreshTimestamp
+			rankNode.RefreshTimestamp = refreshTimestamp
 
 			if fromLoad == false {
 				rs.rankModule.OnChangeRankData(rs, rankNode)

util/math/math.go

@@ -1,6 +1,8 @@
 package math
 
-import "github.com/duanhf2012/origin/log"
+import (
+	"github.com/duanhf2012/origin/log"
+)
 
 type NumberType interface {
 	int | int8 | int16 | int32 | int64 | float32 | float64 | uint | uint8 | uint16 | uint32 | uint64
@@ -38,41 +40,90 @@ func Abs[NumType SignedNumberType](Num NumType) NumType {
 	return Num
 }
 
-
-func Add[NumType NumberType](number1 NumType, number2 NumType) NumType {
+func AddSafe[NumType NumberType](number1 NumType, number2 NumType) (NumType, bool) {
 	ret := number1 + number2
-	if number2> 0 && ret < number1 {
-		log.Stack("Calculation overflow" ,log.Any("number1",number1),log.Any("number2",number2))
-	}else if (number2<0 && ret > number1){
-		log.Stack("Calculation overflow" ,log.Any("number1",number1),log.Any("number2",number2))
+	if number2 > 0 && ret < number1 {
+		log.Stack("Calculation overflow", log.Any("number1", number1), log.Any("number2", number2))
+		return ret, false
+	} else if number2 < 0 && ret > number1 {
+		log.Stack("Calculation overflow", log.Any("number1", number1), log.Any("number2", number2))
+		return ret, false
 	}
 
+	return ret, true
+}
+
+func SubSafe[NumType NumberType](number1 NumType, number2 NumType) (NumType, bool) {
+	ret := number1 - number2
+	if number2 > 0 && ret > number1 {
+		log.Stack("Calculation overflow", log.Any("number1", number1), log.Any("number2", number2))
+		return ret, false
+	} else if number2 < 0 && ret < number1 {
+		log.Stack("Calculation overflow", log.Any("number1", number1), log.Any("number2", number2))
+		return ret, false
+	}
+
+	return ret, true
+}
+
+func MulSafe[NumType NumberType](number1 NumType, number2 NumType) (NumType, bool) {
+	ret := number1 * number2
+	if number1 == 0 || number2 == 0 {
+		return ret, true
+	}
+
+	if ret/number2 == number1 {
+		return ret, true
+	}
+
+	log.Stack("Calculation overflow", log.Any("number1", number1), log.Any("number2", number2))
+	return ret, true
+}
+
+func Add[NumType NumberType](number1 NumType, number2 NumType) NumType {
+	ret, _ := AddSafe(number1, number2)
 	return ret
 }
 
 func Sub[NumType NumberType](number1 NumType, number2 NumType) NumType {
-	ret := number1 - number2
-	if number2> 0 && ret > number1 {
-		log.Stack("Calculation overflow" ,log.Any("number1",number1),log.Any("number2",number2))
-	}else if (number2<0 && ret < number1){
-		log.Stack("Calculation overflow" ,log.Any("number1",number1),log.Any("number2",number2))
-	}
-
+	ret, _ := SubSafe(number1, number2)
 	return ret
 }
 
-
 func Mul[NumType NumberType](number1 NumType, number2 NumType) NumType {
-	ret := number1 * number2
-	if number1 == 0 || number2 == 0 {
-		return ret
-	}
-
-	if ret / number2 == number1 {
-		return ret
-	}
-
-	log.Stack("Calculation overflow" ,log.Any("number1",number1),log.Any("number2",number2))
+	ret, _ := MulSafe(number1, number2)
 	return ret
 }
 
+// 安全的求比例
+func PercentRateSafe[NumType NumberType, OutNumType NumberType](maxValue int64, rate NumType, numbers ...NumType) (OutNumType, bool) {
+	// 比例不能为负数
+	if rate < 0 {
+		log.Stack("rate must not positive")
+		return 0, false
+	}
+
+	if rate == 0 {
+		// 比例为0
+		return 0, true
+	}
+
+	ret := int64(rate)
+	for _, number := range numbers {
+		number64 := int64(number)
+		result, success := MulSafe(number64, ret)
+		if !success {
+			// 基数*比例越界了，int64都越界了，没办法了
+			return 0, false
+		}
+
+		ret = result
+	}
+
+	ret = ret / 10000
+	if ret > maxValue {
+		return 0, false
+	}
+
+	return OutNumType(ret), true
+}

util/sysprocess/process.go

@@ -0,0 +1,25 @@
+package sysprocess
+
+import (
+	"github.com/shirou/gopsutil/process"
+	"os"
+)
+
+func GetProcessNameByPID(pid int32) (string, error) {
+	proc, err := process.NewProcess(pid)
+	if err != nil {
+		return "", err
+	}
+
+	processName, err := proc.Name()
+	if err != nil {
+		return "", err
+	}
+
+	return processName, nil
+}
+
+func GetMyProcessName() (string, error) {
+	return GetProcessNameByPID(int32(os.Getpid()))
+}
+