优化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,
+		"MaxRpcParamLen": 409600
-		"NeighborService":["HttpGateService"]
 	},
 	{
 		"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。其中:
+MasterDiscoveryService: 表示将筛选origin网络中MasterNodeId为2中的TestService8服务，注意如果MasterDiscoveryService不配置，则筛选功能不生效。MasterNodeId也可以填为0，表示NodeId为1的结点，在所有网络中只发现TestService8的服务。
-
-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不配置，则筛选功能不生效。
 
 第八章：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

@@ -20,6 +20,11 @@ const (
 	Discard NodeStatus = 1 //丢弃
 )
 
+type MasterDiscoveryService struct {
+	MasterNodeId int32 //要筛选的主结点Id，如果不配置或者配置成0，表示针对所有的主结点
+	DiscoveryService  []string  //只发现的服务列表
+}
+
 type NodeInfo struct {
 	NodeId            int
 	NodeName          string
@@ -29,9 +34,9 @@ type NodeInfo struct {
 	CompressBytesLen  int   //超过字节进行压缩的长度
 	ServiceList  	  []string //所有的有序服务列表
 	PublicServiceList []string //对外公开的服务列表
-	DiscoveryService  []string //筛选发现的服务，如果不配置，不进行筛选
+	MasterDiscoveryService  []MasterDiscoveryService //筛选发现的服务，如果不配置，不进行筛选
-	NeighborService   []string
 	status            NodeStatus
+	Retire bool
 }
 
 type NodeRpcInfo struct {
@@ -51,8 +56,8 @@ type Cluster struct {
 
 
 	locker         sync.RWMutex                //结点与服务关系保护锁
-	mapRpc           map[int]NodeRpcInfo    //nodeId
+	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]
 
 	rpcServer                rpc.Server
@@ -83,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)
 	}
 }
@@ -99,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]
+	if immediately ==false {
-	for {
-		//立即删除
-		if immediately || ok == false {
-			break
-		}
-
 		//正在连接中不主动断开，只断开没有连接中的
 		if rpc.client.IsConnected() {
-			nodeInfo.status = Discard
+			rpc.nodeInfo.status = Discard
-			log.Info("Discard node",log.Int("nodeId",nodeInfo.NodeId),log.String("ListenAddr", nodeInfo.ListenAddr))
+			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) {
@@ -164,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)
 		}
 	}
@@ -185,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 lastNodeInfo != nil {
-	if _, rpcInfoOK := cls.mapRpc[nodeInfo.NodeId]; rpcInfoOK == true {
+		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.初始化配置
@@ -215,7 +207,6 @@ func (cls *Cluster) Init(localNodeId int, setupServiceFun SetupServiceFun) error
 	}
 
 	cls.rpcServer.Init(cls)
-	cls.buildLocalRpc()
 
 	//2.安装服务发现结点
 	cls.SetupServiceDiscovery(localNodeId, setupServiceFun)
@@ -309,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
 	}
@@ -341,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 {
@@ -349,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]
@@ -471,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]
+	nodeInfo,ok:= cls.mapRpc[nodeId]
-	return nodeInfo,ok
+	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) {
+func (ds *DynamicDiscoveryMaster) updateNodeInfo(nInfo *rpc.NodeInfo) {
-	if len(nodeInfo.PublicServiceList) == 0 {
+	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)
 
@@ -248,15 +280,6 @@ func (dc *DynamicDiscoveryClient) fullCompareDiffNode(masterNodeId int32, mapNod
 
 //订阅发现的服务通知
 func (dc *DynamicDiscoveryClient) RPC_SubServiceDiscover(req *rpc.SubscribeDiscoverNotify) error {
-	//整理当前master结点需要筛选的NeighborService
-	masterDiscoveryNodeInfo := cluster.GetMasterDiscoveryNodeInfo(int(req.MasterNodeId))
-	mapMasterDiscoveryService := map[string]struct{}{}
-	if masterDiscoveryNodeInfo != nil {
-		for i := 0; i < len(masterDiscoveryNodeInfo.NeighborService); i++ {
-			mapMasterDiscoveryService[masterDiscoveryNodeInfo.NeighborService[i]] = struct{}{}
-		}
-	}
-
 	mapNodeInfo := map[int32]*rpc.NodeInfo{}
 	for _, nodeInfo := range req.NodeInfo {
 		//不对本地结点或者不存在任何公开服务的结点
@@ -271,13 +294,6 @@ func (dc *DynamicDiscoveryClient) RPC_SubServiceDiscover(req *rpc.SubscribeDisco
 
 		//遍历所有的公开服务，并筛选之
 		for _, serviceName := range nodeInfo.PublicServiceList {
-			//只有存在配置时才做筛选
-			if len(mapMasterDiscoveryService) > 0 {
-				if _, ok := mapMasterDiscoveryService[serviceName]; ok == false {
-					continue
-				}
-			}
-
 			nInfo := mapNodeInfo[nodeInfo.NodeId]
 			if nInfo == nil {
 				nInfo = &rpc.NodeInfo{}
@@ -285,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
 			}
 
@@ -294,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 {
@@ -309,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),nil)
-		cluster.TriggerDiscoveryEvent(false,int(nodeId),nodeInfo.PublicServiceList)
 		dc.removeMasterNode(req.MasterNodeId, int32(nodeId))
 		if dc.findNodeId(nodeId) == false {
 			dc.funDelService(int(nodeId), false)
@@ -319,10 +336,8 @@ func (dc *DynamicDiscoveryClient) RPC_SubServiceDiscover(req *rpc.SubscribeDisco
 
 	//设置新结点
 	for _, nodeInfo := range mapNodeInfo {
-		dc.addMasterNode(req.MasterNodeId, nodeInfo.NodeId)
+		bSet := dc.setNodeInfo(req.MasterNodeId,nodeInfo)
-		dc.setNodeInfo(nodeInfo)
+		if bSet == false {
-
-		if len(nodeInfo.PublicServiceList) == 0 {
 			continue
 		}
 
@@ -346,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 {
@@ -358,13 +396,9 @@ func (dc *DynamicDiscoveryClient) regServiceDiscover(nodeId int){
 	req.NodeInfo.NodeName = cluster.localNodeInfo.NodeName
 	req.NodeInfo.ListenAddr = cluster.localNodeInfo.ListenAddr
 	req.NodeInfo.MaxRpcParamLen = cluster.localNodeInfo.MaxRpcParamLen
-
+	req.NodeInfo.PublicServiceList =  cluster.localNodeInfo.PublicServiceList
-	//MasterDiscoveryNode配置中没有配置NeighborService，则同步当前结点所有服务
+	req.NodeInfo.Retire = dc.bRetire
-	if len(nodeInfo.NeighborService) == 0 {
+	req.NodeInfo.Private = cluster.localNodeInfo.Private
-		req.NodeInfo.PublicServiceList = cluster.localNodeInfo.PublicServiceList
-	} else {
-		req.NodeInfo.PublicServiceList = append(req.NodeInfo.PublicServiceList, DynamicDiscoveryClientName)
-	}
 
 	//向Master服务同步本Node服务信息
 	err := dc.AsyncCallNode(nodeId, RegServiceDiscover, &req, func(res *rpc.Empty, err error) {
@@ -382,37 +416,56 @@ func (dc *DynamicDiscoveryClient) regServiceDiscover(nodeId int){
 	}
 }
 
-func (dc *DynamicDiscoveryClient) setNodeInfo(nodeInfo *rpc.NodeInfo) {
+func (dc *DynamicDiscoveryClient) canDiscoveryService(fromMasterNodeId int32,serviceName string) bool{
-	if nodeInfo == nil || nodeInfo.Private == true || int(nodeInfo.NodeId) == dc.localNodeId {
+	canDiscovery := true
-		return
-	}
 
-	//筛选关注的服务
+	for i:=0;i<len(cluster.GetLocalNodeInfo().MasterDiscoveryService);i++{
-	localNodeInfo := cluster.GetLocalNodeInfo()
+		masterNodeId := cluster.GetLocalNodeInfo().MasterDiscoveryService[i].MasterNodeId
-	if len(localNodeInfo.DiscoveryService) > 0 {
+
-		var discoverServiceSlice = make([]string, 0, 24)
+		if masterNodeId == fromMasterNodeId || masterNodeId == 0 {
-		for _, pubService := range nodeInfo.PublicServiceList {
+			canDiscovery = false
-			for _, discoverService := range localNodeInfo.DiscoveryService {
+
-				if pubService == discoverService {
+			for _,discoveryService := range cluster.GetLocalNodeInfo().MasterDiscoveryService[i].DiscoveryService {
-					discoverServiceSlice = append(discoverServiceSlice, pubService)
+				if discoveryService == serviceName {
+					return true
 				}
 			}
 		}
-		nodeInfo.PublicServiceList = discoverServiceSlice
 	}
 
-	if len(nodeInfo.PublicServiceList) == 0 {
+	return canDiscovery
-		return
+}
+
+func (dc *DynamicDiscoveryClient) setNodeInfo(masterNodeId int32,nodeInfo *rpc.NodeInfo) bool{
+	if nodeInfo == nil || nodeInfo.Private == true || int(nodeInfo.NodeId) == dc.localNodeId {
+		return false
+	}
+
+	//筛选关注的服务
+	var discoverServiceSlice = make([]string, 0, 24)
+	for _, pubService := range nodeInfo.PublicServiceList {
+		if dc.canDiscoveryService(masterNodeId,pubService) == true {
+			discoverServiceSlice = append(discoverServiceSlice,pubService)
+		}
+	}
+
+	if len(discoverServiceSlice) == 0 {
+		return false
 	}
 
 	var nInfo NodeInfo
-	nInfo.ServiceList = nodeInfo.PublicServiceList
+	nInfo.ServiceList = discoverServiceSlice
-	nInfo.PublicServiceList = nodeInfo.PublicServiceList
+	nInfo.PublicServiceList = discoverServiceSlice
 	nInfo.NodeId = int(nodeInfo.NodeId)
 	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
 }
 
 func (dc *DynamicDiscoveryClient) OnNodeDisconnect(nodeId int) {

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.mapRpc = map[int]*NodeRpcInfo{}
-	cls.mapIdNode = map[int]NodeInfo{}
 	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)
+			pClient,retire := GetCluster().getRpcClient(nodeId)
-			if pClient == nil || (bAll == false && pClient.IsConnected() == false) {
+			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_Tcp         		EventType = -3
-	Sys_Event_Http_Event  			EventType = -4
+	Sys_Event_Http_Event  		EventType = -4
-	Sys_Event_WebSocket       EventType = -5
+	Sys_Event_WebSocket       	EventType = -5
-	Sys_Event_Node_Event      EventType = -6
+	Sys_Event_Node_Event      	EventType = -6
-	Sys_Event_DiscoverService EventType = -7
+	Sys_Event_DiscoverService 	EventType = -7
-	Sys_Event_DiscardGoroutine EventType = -8
+	Sys_Event_DiscardGoroutine 	EventType = -8
-	Sys_Event_QueueTaskFinish  EventType = -9
+	Sys_Event_QueueTaskFinish  	EventType = -9
-	
+	Sys_Event_Retire       	EventType = -10
-	Sys_Event_User_Define EventType = 1
-
 
+	Sys_Event_User_Define 		EventType = 1
 )
 

go.mod

@@ -4,27 +4,34 @@ go 1.21
 
 require (
 	github.com/go-sql-driver/mysql v1.6.0
-	github.com/gogo/protobuf v1.3.2
 	github.com/gomodule/redigo v1.8.8
 	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,25 +1,25 @@
 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/golang/protobuf v1.5.0/go.mod h1:FsONVRAS9T7sI+LIUmWTfcYkHO4aIWwzhcaSAoJOfIk=
-github.com/gogo/protobuf v1.3.2/go.mod h1:P1XiOD3dCwIKUDQYPy72D8LYyHL2YPYrpS2s69NZV8Q=
 github.com/golang/snappy v0.0.1 h1:Qgr9rKW7uDUkrbSmQeiDsGa8SjGyCOGtuasMWwvp2P4=
 github.com/golang/snappy v0.0.1/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q=
 github.com/gomodule/redigo v1.8.8 h1:f6cXq6RRfiyrOJEV7p3JhLDlmawGBVBBP1MggY8Mo4E=
 github.com/gomodule/redigo v1.8.8/go.mod h1:7ArFNvsTjH8GMMzB4uy1snslv2BwmginuMs06a1uzZE=
-github.com/google/go-cmp v0.5.2 h1:X2ev0eStA3AbceY54o37/0PQ/UWqKEiiO2dKL5OPaFM=
 github.com/google/go-cmp v0.5.2/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
+github.com/google/go-cmp v0.5.5 h1:Khx7svrCpmxxtHBq5j2mp/xVjsi8hQMfNLvJFAlrGgU=
+github.com/google/go-cmp v0.5.5/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
 github.com/google/gofuzz v1.0.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg=
 github.com/gorilla/websocket v1.5.0 h1:PPwGk2jz7EePpoHN/+ClbZu8SPxiqlu12wZP/3sWmnc=
 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=
@@ -32,10 +32,14 @@ github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421/go.mod h1:6dJ
 github.com/modern-go/reflect2 v1.0.2 h1:xBagoLtFs94CBntxluKeaWgTMpvLxC4ur3nMaC9Gz0M=
 github.com/modern-go/reflect2 v1.0.2/go.mod h1:yWuevngMOJpCy52FWWMvUC8ws7m/LJsjYzDa0/r8luk=
 github.com/montanaflynn/stats v0.0.0-20171201202039-1bf9dbcd8cbe/go.mod h1:wL8QJuTMNUDYhXwkmfOly8iTdp5TEcJFWZD2D7SIkUc=
+github.com/pierrec/lz4/v4 v4.1.18 h1:xaKrnTkyoqfh1YItXl56+6KJNVYWlEEPuAQW9xsplYQ=
+github.com/pierrec/lz4/v4 v4.1.18/go.mod h1:gZWDp/Ze/IJXGXf23ltt2EXimqmTUXEy0GFuRQyBid4=
 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=
@@ -43,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=
@@ -51,47 +59,37 @@ 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/yusufpapurcu/wmi v1.2.4 h1:zFUKzehAFReQwLys1b/iSMl+JQGSCSjtVqQn9bBrPo0=
-github.com/yuin/goldmark v1.2.1/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
+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/xerrors v0.0.0-20191204190536-9bdfabe68543 h1:E7g+9GITq07hpfrRu66IVDexMakfv52eLZ2CXBWiKr4=
-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/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
-golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1 h1:go1bK/D/BFZV2I8cIQd1NKEZ+0owSTG1fDTci4IqFcE=
+google.golang.org/protobuf v1.26.0-rc.1/go.mod h1:jlhhOSvTdKEhbULTjvd4ARK9grFBp09yW+WbY/TyQbw=
-golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
+google.golang.org/protobuf v1.31.0 h1:g0LDEJHgrBl9N9r17Ru3sqWhkIx2NB67okBHPwC7hs8=
+google.golang.org/protobuf v1.31.0/go.mod h1:HV8QOd/L58Z+nl8r43ehVNZIU/HEI6OcFqwMG9pJV4I=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127 h1:qIbj1fsPNlZgppZ+VLlY7N33q108Sa+fhmuc+sWQYwY=
 gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=

log/handler.go

@@ -1,10 +1,10 @@
 package log
 
 import (
-	"log/slog"
-	"io"
-	"path/filepath"
 	"context"
+	"io"
+	"log/slog"
+	"path/filepath"
 	"runtime"
 	"runtime/debug"
 	"sync"
@@ -35,21 +35,21 @@ type OriginJsonHandler struct {
 func getStrLevel(level slog.Level) string{
 	switch level {
 	case LevelTrace:
-		return "TRACE"
+		return "Trace"
 	case LevelDebug:
-		return "DEBUG"
+		return "Debug"
 	case LevelInfo:
-		return "INFO"
+		return "Info"
 	case LevelWarning:
-		return "WARNING"
+		return "Warning"
 	case LevelError:
-		return "ERROR"
+		return "Error"
 	case LevelStack:
-		return "STACK"
+		return "Stack"
 	case LevelDump:
-		return "DUMP"
+		return "Dump"
 	case LevelFatal:
-		return "FATAL"
+		return "Fatal"
 	}
 
 	return ""
@@ -97,7 +97,8 @@ func (oh *OriginTextHandler) Handle(context context.Context, record slog.Record)
 		oh.w.Write([]byte(strDump))
 		return err
 	}
-	
+
+
 	return oh.TextHandler.Handle(context, record)
 }
 

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 {
+	if OpenConsole {
-		n,err = iw.outConsole.Write(p)
+		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
+// It's non-thread-safe
-func Export(logger *Logger) {
+func SetLogger(logger ILogger) {
-	if logger != nil {
+	if logger != nil && isSetLogger == false {
 		gLogger = logger
+		isSetLogger = true
 	}
 }
 
+func GetLogger() ILogger{
+	return gLogger
+}
+
 func Trace(msg string, args ...any){
 	gLogger.Trace(msg, args...)
 }
@@ -340,6 +357,10 @@ func Close() {
 }
 
 func ErrorAttr(key string,value error) slog.Attr{
+	if value== nil {
+		return slog.Attr{key, slog.StringValue("nil")}
+	}
+
 	return slog.Attr{key, slog.StringValue(value.Error())}
 }
 
@@ -411,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
 	}
@@ -421,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())
@@ -431,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
@@ -485,23 +506,8 @@ func (logger *Logger) formatHeader(buf *Buffer,level slog.Level,calldepth int) {
 	file = filepath.Base(file)
 
 	buf.AppendString("time=\"")
-	year, month, day := t.Date()
+	buf.AppendString(t.Format("2006/01/02 15:04:05"))
-	buf.AppendInt(int64(year))
-	buf.AppendByte('/')
-	buf.AppendInt(int64(month))
-	buf.AppendByte('/')
-	buf.AppendInt(int64(day))
-	buf.AppendByte(' ')
-
-	hour, min, sec := t.Clock()
-	buf.AppendInt(int64(hour))
-	buf.AppendByte(':')
-	buf.AppendInt(int64(min))
-	buf.AppendByte(':')
-	
-	buf.AppendInt(int64(sec))
 	buf.AppendString("\"")
-
 	logger.sBuff.AppendString(" level=")
 	logger.sBuff.AppendString(getStrLevel(level))
 	logger.sBuff.AppendString(" source=")

network/processor/pbprocessor.go

@@ -4,7 +4,7 @@ import (
 	"encoding/binary"
 	"fmt"
 	"github.com/duanhf2012/origin/util/bytespool"
-	"github.com/gogo/protobuf/proto"
+	"google.golang.org/protobuf/proto"
 	"reflect"
 )
 

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)
+	sig = make(chan os.Signal, 4)
-	signal.Notify(sig, syscall.SIGINT, syscall.SIGTERM, syscall.Signal(10))
+	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:
+		case s := <-sig:
-			log.Info("receipt stop signal.")
+			signal := s.(syscall.Signal)
-			bRun = false
+			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.proto

@@ -1,6 +1,6 @@
 syntax = "proto3";
 package rpc;
-option go_package = "./rpc";
+option go_package = ".;rpc";
 
 message NodeInfo{
     int32 NodeId = 1;
@@ -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/gogopbprocessor.go

@@ -1,106 +0,0 @@
-package rpc
-
-import (
-	"github.com/duanhf2012/origin/util/sync"
-	"github.com/gogo/protobuf/proto"
-	"fmt"
-)
-
-type GoGoPBProcessor struct {
-}
-
-var rpcGoGoPbResponseDataPool =sync.NewPool(make(chan interface{},10240), func()interface{}{
-	return &GoGoPBRpcResponseData{}
-})
-
-var rpcGoGoPbRequestDataPool =sync.NewPool(make(chan interface{},10240), func()interface{}{
-	return &GoGoPBRpcRequestData{}
-})
-
-func (slf *GoGoPBRpcRequestData) MakeRequest(seq uint64,rpcMethodId uint32,serviceMethod string,noReply bool,inParam []byte) *GoGoPBRpcRequestData{
-	slf.Seq = seq
-	slf.RpcMethodId = rpcMethodId
-	slf.ServiceMethod = serviceMethod
-	slf.NoReply = noReply
-	slf.InParam = inParam
-
-	return slf
-}
-
-
-func (slf *GoGoPBRpcResponseData) MakeRespone(seq uint64,err RpcError,reply []byte) *GoGoPBRpcResponseData{
-	slf.Seq = seq
-	slf.Error = err.Error()
-	slf.Reply = reply
-
-	return slf
-}
-
-func (slf *GoGoPBProcessor) Marshal(v interface{}) ([]byte, error){
-	return proto.Marshal(v.(proto.Message))
-}
-
-func (slf *GoGoPBProcessor) Unmarshal(data []byte, msg interface{}) error{
-	protoMsg,ok := msg.(proto.Message)
-	if ok == false {
-		return fmt.Errorf("%+v is not of proto.Message type",msg)
-	}
-	return proto.Unmarshal(data, protoMsg)
-}
-
-func (slf *GoGoPBProcessor) MakeRpcRequest(seq uint64,rpcMethodId uint32,serviceMethod string,noReply bool,inParam []byte) IRpcRequestData{
-	pGogoPbRpcRequestData := rpcGoGoPbRequestDataPool.Get().(*GoGoPBRpcRequestData)
-	pGogoPbRpcRequestData.MakeRequest(seq,rpcMethodId,serviceMethod,noReply,inParam)
-	return pGogoPbRpcRequestData
-}
-
-func (slf *GoGoPBProcessor) MakeRpcResponse(seq uint64,err RpcError,reply []byte) IRpcResponseData {
-	pGoGoPBRpcResponseData := rpcGoGoPbResponseDataPool.Get().(*GoGoPBRpcResponseData)
-	pGoGoPBRpcResponseData.MakeRespone(seq,err,reply)
-	return pGoGoPBRpcResponseData
-}
-
-func (slf *GoGoPBProcessor) ReleaseRpcRequest(rpcRequestData IRpcRequestData){
-	rpcGoGoPbRequestDataPool.Put(rpcRequestData)
-}
-
-func (slf *GoGoPBProcessor) ReleaseRpcResponse(rpcResponseData IRpcResponseData){
-	rpcGoGoPbResponseDataPool.Put(rpcResponseData)
-}
-
-func (slf *GoGoPBProcessor) IsParse(param interface{}) bool {
-	_,ok := param.(proto.Message)
-	return ok
-}
-
-func (slf *GoGoPBProcessor)	GetProcessorType() RpcProcessorType{
-	return RpcProcessorGoGoPB
-}
-
-func (slf *GoGoPBProcessor) Clone(src interface{}) (interface{},error){
-	srcMsg,ok := src.(proto.Message)
-	if ok == false {
-		return nil,fmt.Errorf("param is not of proto.message type")
-	}
-
-	return proto.Clone(srcMsg),nil
-}
-
-func (slf *GoGoPBRpcRequestData) IsNoReply() bool{
-	return slf.GetNoReply()
-}
-
-func (slf *GoGoPBRpcResponseData)		GetErr() *RpcError {
-	if slf.GetError() == "" {
-		return nil
-	}
-
-	err := RpcError(slf.GetError())
-	return &err
-}
-
-
-
-
-
-

rpc/gogorpc.pb.go

@@ -1,769 +0,0 @@
-// Code generated by protoc-gen-gogo. DO NOT EDIT.
-// source: gogorpc.proto
-
-package rpc
-
-import (
-	fmt "fmt"
-	proto "github.com/gogo/protobuf/proto"
-	io "io"
-	math "math"
-	math_bits "math/bits"
-)
-
-// Reference imports to suppress errors if they are not otherwise used.
-var _ = proto.Marshal
-var _ = fmt.Errorf
-var _ = math.Inf
-
-// This is a compile-time assertion to ensure that this generated file
-// is compatible with the proto package it is being compiled against.
-// A compilation error at this line likely means your copy of the
-// proto package needs to be updated.
-const _ = proto.GoGoProtoPackageIsVersion3 // please upgrade the proto package
-
-type GoGoPBRpcRequestData struct {
-	Seq                  uint64   `protobuf:"varint,1,opt,name=Seq,proto3" json:"Seq,omitempty"`
-	RpcMethodId          uint32   `protobuf:"varint,2,opt,name=RpcMethodId,proto3" json:"RpcMethodId,omitempty"`
-	ServiceMethod        string   `protobuf:"bytes,3,opt,name=ServiceMethod,proto3" json:"ServiceMethod,omitempty"`
-	NoReply              bool     `protobuf:"varint,4,opt,name=NoReply,proto3" json:"NoReply,omitempty"`
-	InParam              []byte   `protobuf:"bytes,5,opt,name=InParam,proto3" json:"InParam,omitempty"`
-	XXX_NoUnkeyedLiteral struct{} `json:"-"`
-	XXX_unrecognized     []byte   `json:"-"`
-	XXX_sizecache        int32    `json:"-"`
-}
-
-func (m *GoGoPBRpcRequestData) Reset()         { *m = GoGoPBRpcRequestData{} }
-func (m *GoGoPBRpcRequestData) String() string { return proto.CompactTextString(m) }
-func (*GoGoPBRpcRequestData) ProtoMessage()    {}
-func (*GoGoPBRpcRequestData) Descriptor() ([]byte, []int) {
-	return fileDescriptor_d0e25d3af112ec8f, []int{0}
-}
-func (m *GoGoPBRpcRequestData) XXX_Unmarshal(b []byte) error {
-	return m.Unmarshal(b)
-}
-func (m *GoGoPBRpcRequestData) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) {
-	if deterministic {
-		return xxx_messageInfo_GoGoPBRpcRequestData.Marshal(b, m, deterministic)
-	} else {
-		b = b[:cap(b)]
-		n, err := m.MarshalToSizedBuffer(b)
-		if err != nil {
-			return nil, err
-		}
-		return b[:n], nil
-	}
-}
-func (m *GoGoPBRpcRequestData) XXX_Merge(src proto.Message) {
-	xxx_messageInfo_GoGoPBRpcRequestData.Merge(m, src)
-}
-func (m *GoGoPBRpcRequestData) XXX_Size() int {
-	return m.Size()
-}
-func (m *GoGoPBRpcRequestData) XXX_DiscardUnknown() {
-	xxx_messageInfo_GoGoPBRpcRequestData.DiscardUnknown(m)
-}
-
-var xxx_messageInfo_GoGoPBRpcRequestData proto.InternalMessageInfo
-
-func (m *GoGoPBRpcRequestData) GetSeq() uint64 {
-	if m != nil {
-		return m.Seq
-	}
-	return 0
-}
-
-func (m *GoGoPBRpcRequestData) GetRpcMethodId() uint32 {
-	if m != nil {
-		return m.RpcMethodId
-	}
-	return 0
-}
-
-func (m *GoGoPBRpcRequestData) GetServiceMethod() string {
-	if m != nil {
-		return m.ServiceMethod
-	}
-	return ""
-}
-
-func (m *GoGoPBRpcRequestData) GetNoReply() bool {
-	if m != nil {
-		return m.NoReply
-	}
-	return false
-}
-
-func (m *GoGoPBRpcRequestData) GetInParam() []byte {
-	if m != nil {
-		return m.InParam
-	}
-	return nil
-}
-
-type GoGoPBRpcResponseData struct {
-	Seq                  uint64   `protobuf:"varint,1,opt,name=Seq,proto3" json:"Seq,omitempty"`
-	Error                string   `protobuf:"bytes,2,opt,name=Error,proto3" json:"Error,omitempty"`
-	Reply                []byte   `protobuf:"bytes,3,opt,name=Reply,proto3" json:"Reply,omitempty"`
-	XXX_NoUnkeyedLiteral struct{} `json:"-"`
-	XXX_unrecognized     []byte   `json:"-"`
-	XXX_sizecache        int32    `json:"-"`
-}
-
-func (m *GoGoPBRpcResponseData) Reset()         { *m = GoGoPBRpcResponseData{} }
-func (m *GoGoPBRpcResponseData) String() string { return proto.CompactTextString(m) }
-func (*GoGoPBRpcResponseData) ProtoMessage()    {}
-func (*GoGoPBRpcResponseData) Descriptor() ([]byte, []int) {
-	return fileDescriptor_d0e25d3af112ec8f, []int{1}
-}
-func (m *GoGoPBRpcResponseData) XXX_Unmarshal(b []byte) error {
-	return m.Unmarshal(b)
-}
-func (m *GoGoPBRpcResponseData) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) {
-	if deterministic {
-		return xxx_messageInfo_GoGoPBRpcResponseData.Marshal(b, m, deterministic)
-	} else {
-		b = b[:cap(b)]
-		n, err := m.MarshalToSizedBuffer(b)
-		if err != nil {
-			return nil, err
-		}
-		return b[:n], nil
-	}
-}
-func (m *GoGoPBRpcResponseData) XXX_Merge(src proto.Message) {
-	xxx_messageInfo_GoGoPBRpcResponseData.Merge(m, src)
-}
-func (m *GoGoPBRpcResponseData) XXX_Size() int {
-	return m.Size()
-}
-func (m *GoGoPBRpcResponseData) XXX_DiscardUnknown() {
-	xxx_messageInfo_GoGoPBRpcResponseData.DiscardUnknown(m)
-}
-
-var xxx_messageInfo_GoGoPBRpcResponseData proto.InternalMessageInfo
-
-func (m *GoGoPBRpcResponseData) GetSeq() uint64 {
-	if m != nil {
-		return m.Seq
-	}
-	return 0
-}
-
-func (m *GoGoPBRpcResponseData) GetError() string {
-	if m != nil {
-		return m.Error
-	}
-	return ""
-}
-
-func (m *GoGoPBRpcResponseData) GetReply() []byte {
-	if m != nil {
-		return m.Reply
-	}
-	return nil
-}
-
-func init() {
-	proto.RegisterType((*GoGoPBRpcRequestData)(nil), "rpc.GoGoPBRpcRequestData")
-	proto.RegisterType((*GoGoPBRpcResponseData)(nil), "rpc.GoGoPBRpcResponseData")
-}
-
-func init() { proto.RegisterFile("gogorpc.proto", fileDescriptor_d0e25d3af112ec8f) }
-
-var fileDescriptor_d0e25d3af112ec8f = []byte{
-	// 233 bytes of a gzipped FileDescriptorProto
-	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xe2, 0xe2, 0x4d, 0xcf, 0x4f, 0xcf,
-	0x2f, 0x2a, 0x48, 0xd6, 0x2b, 0x28, 0xca, 0x2f, 0xc9, 0x17, 0x62, 0x2e, 0x2a, 0x48, 0x56, 0x5a,
-	0xc2, 0xc8, 0x25, 0xe2, 0x9e, 0xef, 0x9e, 0x1f, 0xe0, 0x14, 0x54, 0x90, 0x1c, 0x94, 0x5a, 0x58,
-	0x9a, 0x5a, 0x5c, 0xe2, 0x92, 0x58, 0x92, 0x28, 0x24, 0xc0, 0xc5, 0x1c, 0x9c, 0x5a, 0x28, 0xc1,
-	0xa8, 0xc0, 0xa8, 0xc1, 0x12, 0x04, 0x62, 0x0a, 0x29, 0x70, 0x71, 0x07, 0x15, 0x24, 0xfb, 0xa6,
-	0x96, 0x64, 0xe4, 0xa7, 0x78, 0xa6, 0x48, 0x30, 0x29, 0x30, 0x6a, 0xf0, 0x06, 0x21, 0x0b, 0x09,
-	0xa9, 0x70, 0xf1, 0x06, 0xa7, 0x16, 0x95, 0x65, 0x26, 0xa7, 0x42, 0x84, 0x24, 0x98, 0x15, 0x18,
-	0x35, 0x38, 0x83, 0x50, 0x05, 0x85, 0x24, 0xb8, 0xd8, 0xfd, 0xf2, 0x83, 0x52, 0x0b, 0x72, 0x2a,
-	0x25, 0x58, 0x14, 0x18, 0x35, 0x38, 0x82, 0x60, 0x5c, 0x90, 0x8c, 0x67, 0x5e, 0x40, 0x62, 0x51,
-	0x62, 0xae, 0x04, 0xab, 0x02, 0xa3, 0x06, 0x4f, 0x10, 0x8c, 0xab, 0x14, 0xca, 0x25, 0x8a, 0xe4,
-	0xca, 0xe2, 0x82, 0xfc, 0xbc, 0xe2, 0x54, 0x1c, 0xce, 0x14, 0xe1, 0x62, 0x75, 0x2d, 0x2a, 0xca,
-	0x2f, 0x02, 0x3b, 0x90, 0x33, 0x08, 0xc2, 0x01, 0x89, 0x42, 0xac, 0x64, 0x06, 0x1b, 0x0c, 0xe1,
-	0x38, 0x09, 0x9f, 0x78, 0x24, 0xc7, 0x78, 0xe1, 0x91, 0x1c, 0xe3, 0x83, 0x47, 0x72, 0x8c, 0x51,
-	0xac, 0x7a, 0xfa, 0x45, 0x05, 0xc9, 0x49, 0x6c, 0xe0, 0xe0, 0x31, 0x06, 0x04, 0x00, 0x00, 0xff,
-	0xff, 0x26, 0xcf, 0x31, 0x39, 0x2f, 0x01, 0x00, 0x00,
-}
-
-func (m *GoGoPBRpcRequestData) Marshal() (dAtA []byte, err error) {
-	size := m.Size()
-	dAtA = make([]byte, size)
-	n, err := m.MarshalToSizedBuffer(dAtA[:size])
-	if err != nil {
-		return nil, err
-	}
-	return dAtA[:n], nil
-}
-
-func (m *GoGoPBRpcRequestData) MarshalTo(dAtA []byte) (int, error) {
-	size := m.Size()
-	return m.MarshalToSizedBuffer(dAtA[:size])
-}
-
-func (m *GoGoPBRpcRequestData) MarshalToSizedBuffer(dAtA []byte) (int, error) {
-	i := len(dAtA)
-	_ = i
-	var l int
-	_ = l
-	if m.XXX_unrecognized != nil {
-		i -= len(m.XXX_unrecognized)
-		copy(dAtA[i:], m.XXX_unrecognized)
-	}
-	if len(m.InParam) > 0 {
-		i -= len(m.InParam)
-		copy(dAtA[i:], m.InParam)
-		i = encodeVarintGogorpc(dAtA, i, uint64(len(m.InParam)))
-		i--
-		dAtA[i] = 0x2a
-	}
-	if m.NoReply {
-		i--
-		if m.NoReply {
-			dAtA[i] = 1
-		} else {
-			dAtA[i] = 0
-		}
-		i--
-		dAtA[i] = 0x20
-	}
-	if len(m.ServiceMethod) > 0 {
-		i -= len(m.ServiceMethod)
-		copy(dAtA[i:], m.ServiceMethod)
-		i = encodeVarintGogorpc(dAtA, i, uint64(len(m.ServiceMethod)))
-		i--
-		dAtA[i] = 0x1a
-	}
-	if m.RpcMethodId != 0 {
-		i = encodeVarintGogorpc(dAtA, i, uint64(m.RpcMethodId))
-		i--
-		dAtA[i] = 0x10
-	}
-	if m.Seq != 0 {
-		i = encodeVarintGogorpc(dAtA, i, uint64(m.Seq))
-		i--
-		dAtA[i] = 0x8
-	}
-	return len(dAtA) - i, nil
-}
-
-func (m *GoGoPBRpcResponseData) Marshal() (dAtA []byte, err error) {
-	size := m.Size()
-	dAtA = make([]byte, size)
-	n, err := m.MarshalToSizedBuffer(dAtA[:size])
-	if err != nil {
-		return nil, err
-	}
-	return dAtA[:n], nil
-}
-
-func (m *GoGoPBRpcResponseData) MarshalTo(dAtA []byte) (int, error) {
-	size := m.Size()
-	return m.MarshalToSizedBuffer(dAtA[:size])
-}
-
-func (m *GoGoPBRpcResponseData) MarshalToSizedBuffer(dAtA []byte) (int, error) {
-	i := len(dAtA)
-	_ = i
-	var l int
-	_ = l
-	if m.XXX_unrecognized != nil {
-		i -= len(m.XXX_unrecognized)
-		copy(dAtA[i:], m.XXX_unrecognized)
-	}
-	if len(m.Reply) > 0 {
-		i -= len(m.Reply)
-		copy(dAtA[i:], m.Reply)
-		i = encodeVarintGogorpc(dAtA, i, uint64(len(m.Reply)))
-		i--
-		dAtA[i] = 0x1a
-	}
-	if len(m.Error) > 0 {
-		i -= len(m.Error)
-		copy(dAtA[i:], m.Error)
-		i = encodeVarintGogorpc(dAtA, i, uint64(len(m.Error)))
-		i--
-		dAtA[i] = 0x12
-	}
-	if m.Seq != 0 {
-		i = encodeVarintGogorpc(dAtA, i, uint64(m.Seq))
-		i--
-		dAtA[i] = 0x8
-	}
-	return len(dAtA) - i, nil
-}
-
-func encodeVarintGogorpc(dAtA []byte, offset int, v uint64) int {
-	offset -= sovGogorpc(v)
-	base := offset
-	for v >= 1<<7 {
-		dAtA[offset] = uint8(v&0x7f | 0x80)
-		v >>= 7
-		offset++
-	}
-	dAtA[offset] = uint8(v)
-	return base
-}
-func (m *GoGoPBRpcRequestData) Size() (n int) {
-	if m == nil {
-		return 0
-	}
-	var l int
-	_ = l
-	if m.Seq != 0 {
-		n += 1 + sovGogorpc(uint64(m.Seq))
-	}
-	if m.RpcMethodId != 0 {
-		n += 1 + sovGogorpc(uint64(m.RpcMethodId))
-	}
-	l = len(m.ServiceMethod)
-	if l > 0 {
-		n += 1 + l + sovGogorpc(uint64(l))
-	}
-	if m.NoReply {
-		n += 2
-	}
-	l = len(m.InParam)
-	if l > 0 {
-		n += 1 + l + sovGogorpc(uint64(l))
-	}
-	if m.XXX_unrecognized != nil {
-		n += len(m.XXX_unrecognized)
-	}
-	return n
-}
-
-func (m *GoGoPBRpcResponseData) Size() (n int) {
-	if m == nil {
-		return 0
-	}
-	var l int
-	_ = l
-	if m.Seq != 0 {
-		n += 1 + sovGogorpc(uint64(m.Seq))
-	}
-	l = len(m.Error)
-	if l > 0 {
-		n += 1 + l + sovGogorpc(uint64(l))
-	}
-	l = len(m.Reply)
-	if l > 0 {
-		n += 1 + l + sovGogorpc(uint64(l))
-	}
-	if m.XXX_unrecognized != nil {
-		n += len(m.XXX_unrecognized)
-	}
-	return n
-}
-
-func sovGogorpc(x uint64) (n int) {
-	return (math_bits.Len64(x|1) + 6) / 7
-}
-func sozGogorpc(x uint64) (n int) {
-	return sovGogorpc(uint64((x << 1) ^ uint64((int64(x) >> 63))))
-}
-func (m *GoGoPBRpcRequestData) Unmarshal(dAtA []byte) error {
-	l := len(dAtA)
-	iNdEx := 0
-	for iNdEx < l {
-		preIndex := iNdEx
-		var wire uint64
-		for shift := uint(0); ; shift += 7 {
-			if shift >= 64 {
-				return ErrIntOverflowGogorpc
-			}
-			if iNdEx >= l {
-				return io.ErrUnexpectedEOF
-			}
-			b := dAtA[iNdEx]
-			iNdEx++
-			wire |= uint64(b&0x7F) << shift
-			if b < 0x80 {
-				break
-			}
-		}
-		fieldNum := int32(wire >> 3)
-		wireType := int(wire & 0x7)
-		if wireType == 4 {
-			return fmt.Errorf("proto: GoGoPBRpcRequestData: wiretype end group for non-group")
-		}
-		if fieldNum <= 0 {
-			return fmt.Errorf("proto: GoGoPBRpcRequestData: illegal tag %d (wire type %d)", fieldNum, wire)
-		}
-		switch fieldNum {
-		case 1:
-			if wireType != 0 {
-				return fmt.Errorf("proto: wrong wireType = %d for field Seq", wireType)
-			}
-			m.Seq = 0
-			for shift := uint(0); ; shift += 7 {
-				if shift >= 64 {
-					return ErrIntOverflowGogorpc
-				}
-				if iNdEx >= l {
-					return io.ErrUnexpectedEOF
-				}
-				b := dAtA[iNdEx]
-				iNdEx++
-				m.Seq |= uint64(b&0x7F) << shift
-				if b < 0x80 {
-					break
-				}
-			}
-		case 2:
-			if wireType != 0 {
-				return fmt.Errorf("proto: wrong wireType = %d for field RpcMethodId", wireType)
-			}
-			m.RpcMethodId = 0
-			for shift := uint(0); ; shift += 7 {
-				if shift >= 64 {
-					return ErrIntOverflowGogorpc
-				}
-				if iNdEx >= l {
-					return io.ErrUnexpectedEOF
-				}
-				b := dAtA[iNdEx]
-				iNdEx++
-				m.RpcMethodId |= uint32(b&0x7F) << shift
-				if b < 0x80 {
-					break
-				}
-			}
-		case 3:
-			if wireType != 2 {
-				return fmt.Errorf("proto: wrong wireType = %d for field ServiceMethod", wireType)
-			}
-			var stringLen uint64
-			for shift := uint(0); ; shift += 7 {
-				if shift >= 64 {
-					return ErrIntOverflowGogorpc
-				}
-				if iNdEx >= l {
-					return io.ErrUnexpectedEOF
-				}
-				b := dAtA[iNdEx]
-				iNdEx++
-				stringLen |= uint64(b&0x7F) << shift
-				if b < 0x80 {
-					break
-				}
-			}
-			intStringLen := int(stringLen)
-			if intStringLen < 0 {
-				return ErrInvalidLengthGogorpc
-			}
-			postIndex := iNdEx + intStringLen
-			if postIndex < 0 {
-				return ErrInvalidLengthGogorpc
-			}
-			if postIndex > l {
-				return io.ErrUnexpectedEOF
-			}
-			m.ServiceMethod = string(dAtA[iNdEx:postIndex])
-			iNdEx = postIndex
-		case 4:
-			if wireType != 0 {
-				return fmt.Errorf("proto: wrong wireType = %d for field NoReply", wireType)
-			}
-			var v int
-			for shift := uint(0); ; shift += 7 {
-				if shift >= 64 {
-					return ErrIntOverflowGogorpc
-				}
-				if iNdEx >= l {
-					return io.ErrUnexpectedEOF
-				}
-				b := dAtA[iNdEx]
-				iNdEx++
-				v |= int(b&0x7F) << shift
-				if b < 0x80 {
-					break
-				}
-			}
-			m.NoReply = bool(v != 0)
-		case 5:
-			if wireType != 2 {
-				return fmt.Errorf("proto: wrong wireType = %d for field InParam", wireType)
-			}
-			var byteLen int
-			for shift := uint(0); ; shift += 7 {
-				if shift >= 64 {
-					return ErrIntOverflowGogorpc
-				}
-				if iNdEx >= l {
-					return io.ErrUnexpectedEOF
-				}
-				b := dAtA[iNdEx]
-				iNdEx++
-				byteLen |= int(b&0x7F) << shift
-				if b < 0x80 {
-					break
-				}
-			}
-			if byteLen < 0 {
-				return ErrInvalidLengthGogorpc
-			}
-			postIndex := iNdEx + byteLen
-			if postIndex < 0 {
-				return ErrInvalidLengthGogorpc
-			}
-			if postIndex > l {
-				return io.ErrUnexpectedEOF
-			}
-			m.InParam = append(m.InParam[:0], dAtA[iNdEx:postIndex]...)
-			if m.InParam == nil {
-				m.InParam = []byte{}
-			}
-			iNdEx = postIndex
-		default:
-			iNdEx = preIndex
-			skippy, err := skipGogorpc(dAtA[iNdEx:])
-			if err != nil {
-				return err
-			}
-			if skippy < 0 {
-				return ErrInvalidLengthGogorpc
-			}
-			if (iNdEx + skippy) < 0 {
-				return ErrInvalidLengthGogorpc
-			}
-			if (iNdEx + skippy) > l {
-				return io.ErrUnexpectedEOF
-			}
-			m.XXX_unrecognized = append(m.XXX_unrecognized, dAtA[iNdEx:iNdEx+skippy]...)
-			iNdEx += skippy
-		}
-	}
-
-	if iNdEx > l {
-		return io.ErrUnexpectedEOF
-	}
-	return nil
-}
-func (m *GoGoPBRpcResponseData) Unmarshal(dAtA []byte) error {
-	l := len(dAtA)
-	iNdEx := 0
-	for iNdEx < l {
-		preIndex := iNdEx
-		var wire uint64
-		for shift := uint(0); ; shift += 7 {
-			if shift >= 64 {
-				return ErrIntOverflowGogorpc
-			}
-			if iNdEx >= l {
-				return io.ErrUnexpectedEOF
-			}
-			b := dAtA[iNdEx]
-			iNdEx++
-			wire |= uint64(b&0x7F) << shift
-			if b < 0x80 {
-				break
-			}
-		}
-		fieldNum := int32(wire >> 3)
-		wireType := int(wire & 0x7)
-		if wireType == 4 {
-			return fmt.Errorf("proto: GoGoPBRpcResponseData: wiretype end group for non-group")
-		}
-		if fieldNum <= 0 {
-			return fmt.Errorf("proto: GoGoPBRpcResponseData: illegal tag %d (wire type %d)", fieldNum, wire)
-		}
-		switch fieldNum {
-		case 1:
-			if wireType != 0 {
-				return fmt.Errorf("proto: wrong wireType = %d for field Seq", wireType)
-			}
-			m.Seq = 0
-			for shift := uint(0); ; shift += 7 {
-				if shift >= 64 {
-					return ErrIntOverflowGogorpc
-				}
-				if iNdEx >= l {
-					return io.ErrUnexpectedEOF
-				}
-				b := dAtA[iNdEx]
-				iNdEx++
-				m.Seq |= uint64(b&0x7F) << shift
-				if b < 0x80 {
-					break
-				}
-			}
-		case 2:
-			if wireType != 2 {
-				return fmt.Errorf("proto: wrong wireType = %d for field Error", wireType)
-			}
-			var stringLen uint64
-			for shift := uint(0); ; shift += 7 {
-				if shift >= 64 {
-					return ErrIntOverflowGogorpc
-				}
-				if iNdEx >= l {
-					return io.ErrUnexpectedEOF
-				}
-				b := dAtA[iNdEx]
-				iNdEx++
-				stringLen |= uint64(b&0x7F) << shift
-				if b < 0x80 {
-					break
-				}
-			}
-			intStringLen := int(stringLen)
-			if intStringLen < 0 {
-				return ErrInvalidLengthGogorpc
-			}
-			postIndex := iNdEx + intStringLen
-			if postIndex < 0 {
-				return ErrInvalidLengthGogorpc
-			}
-			if postIndex > l {
-				return io.ErrUnexpectedEOF
-			}
-			m.Error = string(dAtA[iNdEx:postIndex])
-			iNdEx = postIndex
-		case 3:
-			if wireType != 2 {
-				return fmt.Errorf("proto: wrong wireType = %d for field Reply", wireType)
-			}
-			var byteLen int
-			for shift := uint(0); ; shift += 7 {
-				if shift >= 64 {
-					return ErrIntOverflowGogorpc
-				}
-				if iNdEx >= l {
-					return io.ErrUnexpectedEOF
-				}
-				b := dAtA[iNdEx]
-				iNdEx++
-				byteLen |= int(b&0x7F) << shift
-				if b < 0x80 {
-					break
-				}
-			}
-			if byteLen < 0 {
-				return ErrInvalidLengthGogorpc
-			}
-			postIndex := iNdEx + byteLen
-			if postIndex < 0 {
-				return ErrInvalidLengthGogorpc
-			}
-			if postIndex > l {
-				return io.ErrUnexpectedEOF
-			}
-			m.Reply = append(m.Reply[:0], dAtA[iNdEx:postIndex]...)
-			if m.Reply == nil {
-				m.Reply = []byte{}
-			}
-			iNdEx = postIndex
-		default:
-			iNdEx = preIndex
-			skippy, err := skipGogorpc(dAtA[iNdEx:])
-			if err != nil {
-				return err
-			}
-			if skippy < 0 {
-				return ErrInvalidLengthGogorpc
-			}
-			if (iNdEx + skippy) < 0 {
-				return ErrInvalidLengthGogorpc
-			}
-			if (iNdEx + skippy) > l {
-				return io.ErrUnexpectedEOF
-			}
-			m.XXX_unrecognized = append(m.XXX_unrecognized, dAtA[iNdEx:iNdEx+skippy]...)
-			iNdEx += skippy
-		}
-	}
-
-	if iNdEx > l {
-		return io.ErrUnexpectedEOF
-	}
-	return nil
-}
-func skipGogorpc(dAtA []byte) (n int, err error) {
-	l := len(dAtA)
-	iNdEx := 0
-	depth := 0
-	for iNdEx < l {
-		var wire uint64
-		for shift := uint(0); ; shift += 7 {
-			if shift >= 64 {
-				return 0, ErrIntOverflowGogorpc
-			}
-			if iNdEx >= l {
-				return 0, io.ErrUnexpectedEOF
-			}
-			b := dAtA[iNdEx]
-			iNdEx++
-			wire |= (uint64(b) & 0x7F) << shift
-			if b < 0x80 {
-				break
-			}
-		}
-		wireType := int(wire & 0x7)
-		switch wireType {
-		case 0:
-			for shift := uint(0); ; shift += 7 {
-				if shift >= 64 {
-					return 0, ErrIntOverflowGogorpc
-				}
-				if iNdEx >= l {
-					return 0, io.ErrUnexpectedEOF
-				}
-				iNdEx++
-				if dAtA[iNdEx-1] < 0x80 {
-					break
-				}
-			}
-		case 1:
-			iNdEx += 8
-		case 2:
-			var length int
-			for shift := uint(0); ; shift += 7 {
-				if shift >= 64 {
-					return 0, ErrIntOverflowGogorpc
-				}
-				if iNdEx >= l {
-					return 0, io.ErrUnexpectedEOF
-				}
-				b := dAtA[iNdEx]
-				iNdEx++
-				length |= (int(b) & 0x7F) << shift
-				if b < 0x80 {
-					break
-				}
-			}
-			if length < 0 {
-				return 0, ErrInvalidLengthGogorpc
-			}
-			iNdEx += length
-		case 3:
-			depth++
-		case 4:
-			if depth == 0 {
-				return 0, ErrUnexpectedEndOfGroupGogorpc
-			}
-			depth--
-		case 5:
-			iNdEx += 4
-		default:
-			return 0, fmt.Errorf("proto: illegal wireType %d", wireType)
-		}
-		if iNdEx < 0 {
-			return 0, ErrInvalidLengthGogorpc
-		}
-		if depth == 0 {
-			return iNdEx, nil
-		}
-	}
-	return 0, io.ErrUnexpectedEOF
-}
-
-var (
-	ErrInvalidLengthGogorpc        = fmt.Errorf("proto: negative length found during unmarshaling")
-	ErrIntOverflowGogorpc          = fmt.Errorf("proto: integer overflow")
-	ErrUnexpectedEndOfGroupGogorpc = fmt.Errorf("proto: unexpected end of group")
-)

rpc/pbprocessor.go

@@ -0,0 +1,106 @@
+package rpc
+
+import (
+	"github.com/duanhf2012/origin/util/sync"
+	"google.golang.org/protobuf/proto"
+	"fmt"
+)
+
+type PBProcessor struct {
+}
+
+var rpcPbResponseDataPool =sync.NewPool(make(chan interface{},10240), func()interface{}{
+	return &PBRpcResponseData{}
+})
+
+var rpcPbRequestDataPool =sync.NewPool(make(chan interface{},10240), func()interface{}{
+	return &PBRpcRequestData{}
+})
+
+func (slf *PBRpcRequestData) MakeRequest(seq uint64,rpcMethodId uint32,serviceMethod string,noReply bool,inParam []byte) *PBRpcRequestData{
+	slf.Seq = seq
+	slf.RpcMethodId = rpcMethodId
+	slf.ServiceMethod = serviceMethod
+	slf.NoReply = noReply
+	slf.InParam = inParam
+
+	return slf
+}
+
+
+func (slf *PBRpcResponseData) MakeRespone(seq uint64,err RpcError,reply []byte) *PBRpcResponseData{
+	slf.Seq = seq
+	slf.Error = err.Error()
+	slf.Reply = reply
+
+	return slf
+}
+
+func (slf *PBProcessor) Marshal(v interface{}) ([]byte, error){
+	return proto.Marshal(v.(proto.Message))
+}
+
+func (slf *PBProcessor) Unmarshal(data []byte, msg interface{}) error{
+	protoMsg,ok := msg.(proto.Message)
+	if ok == false {
+		return fmt.Errorf("%+v is not of proto.Message type",msg)
+	}
+	return proto.Unmarshal(data, protoMsg)
+}
+
+func (slf *PBProcessor) MakeRpcRequest(seq uint64,rpcMethodId uint32,serviceMethod string,noReply bool,inParam []byte) IRpcRequestData{
+	pGogoPbRpcRequestData := rpcPbRequestDataPool.Get().(*PBRpcRequestData)
+	pGogoPbRpcRequestData.MakeRequest(seq,rpcMethodId,serviceMethod,noReply,inParam)
+	return pGogoPbRpcRequestData
+}
+
+func (slf *PBProcessor) MakeRpcResponse(seq uint64,err RpcError,reply []byte) IRpcResponseData {
+	pPBRpcResponseData := rpcPbResponseDataPool.Get().(*PBRpcResponseData)
+	pPBRpcResponseData.MakeRespone(seq,err,reply)
+	return pPBRpcResponseData
+}
+
+func (slf *PBProcessor) ReleaseRpcRequest(rpcRequestData IRpcRequestData){
+	rpcPbRequestDataPool.Put(rpcRequestData)
+}
+
+func (slf *PBProcessor) ReleaseRpcResponse(rpcResponseData IRpcResponseData){
+	rpcPbResponseDataPool.Put(rpcResponseData)
+}
+
+func (slf *PBProcessor) IsParse(param interface{}) bool {
+	_,ok := param.(proto.Message)
+	return ok
+}
+
+func (slf *PBProcessor)	GetProcessorType() RpcProcessorType{
+	return RpcProcessorPB
+}
+
+func (slf *PBProcessor) Clone(src interface{}) (interface{},error){
+	srcMsg,ok := src.(proto.Message)
+	if ok == false {
+		return nil,fmt.Errorf("param is not of proto.message type")
+	}
+
+	return proto.Clone(srcMsg),nil
+}
+
+func (slf *PBRpcRequestData) IsNoReply() bool{
+	return slf.GetNoReply()
+}
+
+func (slf *PBRpcResponseData)		GetErr() *RpcError {
+	if slf.GetError() == "" {
+		return nil
+	}
+
+	err := RpcError(slf.GetError())
+	return &err
+}
+
+
+
+
+
+

rpc/protorpc.pb.go

@@ -0,0 +1,263 @@
+// Code generated by protoc-gen-go. DO NOT EDIT.
+// versions:
+// 	protoc-gen-go v1.31.0
+// 	protoc        v3.11.4
+// source: test/rpc/protorpc.proto
+
+package rpc
+
+import (
+	protoreflect "google.golang.org/protobuf/reflect/protoreflect"
+	protoimpl "google.golang.org/protobuf/runtime/protoimpl"
+	reflect "reflect"
+	sync "sync"
+)
+
+const (
+	// Verify that this generated code is sufficiently up-to-date.
+	_ = protoimpl.EnforceVersion(20 - protoimpl.MinVersion)
+	// Verify that runtime/protoimpl is sufficiently up-to-date.
+	_ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20)
+)
+
+type PBRpcRequestData struct {
+	state         protoimpl.MessageState
+	sizeCache     protoimpl.SizeCache
+	unknownFields protoimpl.UnknownFields
+
+	Seq           uint64 `protobuf:"varint,1,opt,name=Seq,proto3" json:"Seq,omitempty"`
+	RpcMethodId   uint32 `protobuf:"varint,2,opt,name=RpcMethodId,proto3" json:"RpcMethodId,omitempty"`
+	ServiceMethod string `protobuf:"bytes,3,opt,name=ServiceMethod,proto3" json:"ServiceMethod,omitempty"`
+	NoReply       bool   `protobuf:"varint,4,opt,name=NoReply,proto3" json:"NoReply,omitempty"`
+	InParam       []byte `protobuf:"bytes,5,opt,name=InParam,proto3" json:"InParam,omitempty"`
+}
+
+func (x *PBRpcRequestData) Reset() {
+	*x = PBRpcRequestData{}
+	if protoimpl.UnsafeEnabled {
+		mi := &file_test_rpc_protorpc_proto_msgTypes[0]
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		ms.StoreMessageInfo(mi)
+	}
+}
+
+func (x *PBRpcRequestData) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*PBRpcRequestData) ProtoMessage() {}
+
+func (x *PBRpcRequestData) ProtoReflect() protoreflect.Message {
+	mi := &file_test_rpc_protorpc_proto_msgTypes[0]
+	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 PBRpcRequestData.ProtoReflect.Descriptor instead.
+func (*PBRpcRequestData) Descriptor() ([]byte, []int) {
+	return file_test_rpc_protorpc_proto_rawDescGZIP(), []int{0}
+}
+
+func (x *PBRpcRequestData) GetSeq() uint64 {
+	if x != nil {
+		return x.Seq
+	}
+	return 0
+}
+
+func (x *PBRpcRequestData) GetRpcMethodId() uint32 {
+	if x != nil {
+		return x.RpcMethodId
+	}
+	return 0
+}
+
+func (x *PBRpcRequestData) GetServiceMethod() string {
+	if x != nil {
+		return x.ServiceMethod
+	}
+	return ""
+}
+
+func (x *PBRpcRequestData) GetNoReply() bool {
+	if x != nil {
+		return x.NoReply
+	}
+	return false
+}
+
+func (x *PBRpcRequestData) GetInParam() []byte {
+	if x != nil {
+		return x.InParam
+	}
+	return nil
+}
+
+type PBRpcResponseData struct {
+	state         protoimpl.MessageState
+	sizeCache     protoimpl.SizeCache
+	unknownFields protoimpl.UnknownFields
+
+	Seq   uint64 `protobuf:"varint,1,opt,name=Seq,proto3" json:"Seq,omitempty"`
+	Error string `protobuf:"bytes,2,opt,name=Error,proto3" json:"Error,omitempty"`
+	Reply []byte `protobuf:"bytes,3,opt,name=Reply,proto3" json:"Reply,omitempty"`
+}
+
+func (x *PBRpcResponseData) Reset() {
+	*x = PBRpcResponseData{}
+	if protoimpl.UnsafeEnabled {
+		mi := &file_test_rpc_protorpc_proto_msgTypes[1]
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		ms.StoreMessageInfo(mi)
+	}
+}
+
+func (x *PBRpcResponseData) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*PBRpcResponseData) ProtoMessage() {}
+
+func (x *PBRpcResponseData) ProtoReflect() protoreflect.Message {
+	mi := &file_test_rpc_protorpc_proto_msgTypes[1]
+	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 PBRpcResponseData.ProtoReflect.Descriptor instead.
+func (*PBRpcResponseData) Descriptor() ([]byte, []int) {
+	return file_test_rpc_protorpc_proto_rawDescGZIP(), []int{1}
+}
+
+func (x *PBRpcResponseData) GetSeq() uint64 {
+	if x != nil {
+		return x.Seq
+	}
+	return 0
+}
+
+func (x *PBRpcResponseData) GetError() string {
+	if x != nil {
+		return x.Error
+	}
+	return ""
+}
+
+func (x *PBRpcResponseData) GetReply() []byte {
+	if x != nil {
+		return x.Reply
+	}
+	return nil
+}
+
+var File_test_rpc_protorpc_proto protoreflect.FileDescriptor
+
+var file_test_rpc_protorpc_proto_rawDesc = []byte{
+	0x0a, 0x17, 0x74, 0x65, 0x73, 0x74, 0x2f, 0x72, 0x70, 0x63, 0x2f, 0x70, 0x72, 0x6f, 0x74, 0x6f,
+	0x72, 0x70, 0x63, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x12, 0x03, 0x72, 0x70, 0x63, 0x22, 0xa0,
+	0x01, 0x0a, 0x10, 0x50, 0x42, 0x52, 0x70, 0x63, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x44,
+	0x61, 0x74, 0x61, 0x12, 0x10, 0x0a, 0x03, 0x53, 0x65, 0x71, 0x18, 0x01, 0x20, 0x01, 0x28, 0x04,
+	0x52, 0x03, 0x53, 0x65, 0x71, 0x12, 0x20, 0x0a, 0x0b, 0x52, 0x70, 0x63, 0x4d, 0x65, 0x74, 0x68,
+	0x6f, 0x64, 0x49, 0x64, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0d, 0x52, 0x0b, 0x52, 0x70, 0x63, 0x4d,
+	0x65, 0x74, 0x68, 0x6f, 0x64, 0x49, 0x64, 0x12, 0x24, 0x0a, 0x0d, 0x53, 0x65, 0x72, 0x76, 0x69,
+	0x63, 0x65, 0x4d, 0x65, 0x74, 0x68, 0x6f, 0x64, 0x18, 0x03, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0d,
+	0x53, 0x65, 0x72, 0x76, 0x69, 0x63, 0x65, 0x4d, 0x65, 0x74, 0x68, 0x6f, 0x64, 0x12, 0x18, 0x0a,
+	0x07, 0x4e, 0x6f, 0x52, 0x65, 0x70, 0x6c, 0x79, 0x18, 0x04, 0x20, 0x01, 0x28, 0x08, 0x52, 0x07,
+	0x4e, 0x6f, 0x52, 0x65, 0x70, 0x6c, 0x79, 0x12, 0x18, 0x0a, 0x07, 0x49, 0x6e, 0x50, 0x61, 0x72,
+	0x61, 0x6d, 0x18, 0x05, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x07, 0x49, 0x6e, 0x50, 0x61, 0x72, 0x61,
+	0x6d, 0x22, 0x51, 0x0a, 0x11, 0x50, 0x42, 0x52, 0x70, 0x63, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e,
+	0x73, 0x65, 0x44, 0x61, 0x74, 0x61, 0x12, 0x10, 0x0a, 0x03, 0x53, 0x65, 0x71, 0x18, 0x01, 0x20,
+	0x01, 0x28, 0x04, 0x52, 0x03, 0x53, 0x65, 0x71, 0x12, 0x14, 0x0a, 0x05, 0x45, 0x72, 0x72, 0x6f,
+	0x72, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x05, 0x45, 0x72, 0x72, 0x6f, 0x72, 0x12, 0x14,
+	0x0a, 0x05, 0x52, 0x65, 0x70, 0x6c, 0x79, 0x18, 0x03, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x05, 0x52,
+	0x65, 0x70, 0x6c, 0x79, 0x42, 0x07, 0x5a, 0x05, 0x2e, 0x3b, 0x72, 0x70, 0x63, 0x62, 0x06, 0x70,
+	0x72, 0x6f, 0x74, 0x6f, 0x33,
+}
+
+var (
+	file_test_rpc_protorpc_proto_rawDescOnce sync.Once
+	file_test_rpc_protorpc_proto_rawDescData = file_test_rpc_protorpc_proto_rawDesc
+)
+
+func file_test_rpc_protorpc_proto_rawDescGZIP() []byte {
+	file_test_rpc_protorpc_proto_rawDescOnce.Do(func() {
+		file_test_rpc_protorpc_proto_rawDescData = protoimpl.X.CompressGZIP(file_test_rpc_protorpc_proto_rawDescData)
+	})
+	return file_test_rpc_protorpc_proto_rawDescData
+}
+
+var file_test_rpc_protorpc_proto_msgTypes = make([]protoimpl.MessageInfo, 2)
+var file_test_rpc_protorpc_proto_goTypes = []interface{}{
+	(*PBRpcRequestData)(nil),  // 0: rpc.PBRpcRequestData
+	(*PBRpcResponseData)(nil), // 1: rpc.PBRpcResponseData
+}
+var file_test_rpc_protorpc_proto_depIdxs = []int32{
+	0, // [0:0] is the sub-list for method output_type
+	0, // [0:0] is the sub-list for method input_type
+	0, // [0:0] is the sub-list for extension type_name
+	0, // [0:0] is the sub-list for extension extendee
+	0, // [0:0] is the sub-list for field type_name
+}
+
+func init() { file_test_rpc_protorpc_proto_init() }
+func file_test_rpc_protorpc_proto_init() {
+	if File_test_rpc_protorpc_proto != nil {
+		return
+	}
+	if !protoimpl.UnsafeEnabled {
+		file_test_rpc_protorpc_proto_msgTypes[0].Exporter = func(v interface{}, i int) interface{} {
+			switch v := v.(*PBRpcRequestData); i {
+			case 0:
+				return &v.state
+			case 1:
+				return &v.sizeCache
+			case 2:
+				return &v.unknownFields
+			default:
+				return nil
+			}
+		}
+		file_test_rpc_protorpc_proto_msgTypes[1].Exporter = func(v interface{}, i int) interface{} {
+			switch v := v.(*PBRpcResponseData); i {
+			case 0:
+				return &v.state
+			case 1:
+				return &v.sizeCache
+			case 2:
+				return &v.unknownFields
+			default:
+				return nil
+			}
+		}
+	}
+	type x struct{}
+	out := protoimpl.TypeBuilder{
+		File: protoimpl.DescBuilder{
+			GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
+			RawDescriptor: file_test_rpc_protorpc_proto_rawDesc,
+			NumEnums:      0,
+			NumMessages:   2,
+			NumExtensions: 0,
+			NumServices:   0,
+		},
+		GoTypes:           file_test_rpc_protorpc_proto_goTypes,
+		DependencyIndexes: file_test_rpc_protorpc_proto_depIdxs,
+		MessageInfos:      file_test_rpc_protorpc_proto_msgTypes,
+	}.Build()
+	File_test_rpc_protorpc_proto = out.File
+	file_test_rpc_protorpc_proto_rawDesc = nil
+	file_test_rpc_protorpc_proto_goTypes = nil
+	file_test_rpc_protorpc_proto_depIdxs = nil
+}

rpc/protorpc.proto

@@ -1,8 +1,8 @@
 syntax = "proto3";
 package rpc;
-option go_package = "./rpc";
+option go_package = ".;rpc";
 
-message GoGoPBRpcRequestData{
+message PBRpcRequestData{
   uint64 Seq            = 1;
   uint32 RpcMethodId    = 2;
   string ServiceMethod  = 3;
@@ -10,7 +10,7 @@ message GoGoPBRpcRequestData{
   bytes  InParam        = 5;
 }
 
-message GoGoPBRpcResponseData{
+message PBRpcResponseData{
   uint64 Seq = 1;
   string Error = 2;
   bytes Reply = 3;

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

@@ -10,16 +10,17 @@ import (
 	"reflect"
 	"strings"
 	"time"
+	"runtime"
 )
 
 type RpcProcessorType uint8
 
 const (
 	RpcProcessorJson   RpcProcessorType = 0
-	RpcProcessorGoGoPB RpcProcessorType = 1
+	RpcProcessorPB RpcProcessorType = 1
 )
 
-var arrayProcessor = []IRpcProcessor{&JsonProcessor{}, &GoGoPBProcessor{}}
+var arrayProcessor = []IRpcProcessor{&JsonProcessor{}, &PBProcessor{}}
 var arrayProcessorLen uint8 = 2
 var LittleEndian bool
 
@@ -141,6 +142,15 @@ func (agent *RpcAgent) WriteResponse(processor IRpcProcessor, serviceMethod stri
 }
 
 func (agent *RpcAgent) Run() {
+	defer func() {
+		if r := recover(); r != nil {
+			buf := make([]byte, 4096)
+			l := runtime.Stack(buf, false) 
+			errString := fmt.Sprint(r)
+			log.Dump(string(buf[:l]),log.String("error",errString))
+		}
+	}()
+
 	for {
 		data, err := agent.conn.ReadMsg()
 		if err != nil {
@@ -226,12 +236,13 @@ func (agent *RpcAgent) Run() {
 		req.inParam, err = rpcHandler.UnmarshalInParam(req.rpcProcessor, req.RpcRequestData.GetServiceMethod(), req.RpcRequestData.GetRpcMethodId(), req.RpcRequestData.GetInParam())
 		if err != nil {
 			rErr := "Call Rpc " + req.RpcRequestData.GetServiceMethod() + " Param error " + err.Error()
+			log.Error("call rpc param error",log.String("serviceMethod",req.RpcRequestData.GetServiceMethod()),log.ErrorAttr("error",err))
 			if req.requestHandle != nil {
 				req.requestHandle(nil, RpcError(rErr))
 			} else {
 				ReleaseRpcRequest(req)
 			}
-			log.Error("call rpc param error",log.String("serviceMethod",req.RpcRequestData.GetServiceMethod()),log.ErrorAttr("error",err))
+
 			continue
 		}
 

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,
+				Timeout:   dialTimeout,
-				KeepAlive: 30 * time.Second,
+				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)

sysservice/tcpservice/tcpservice.go

@@ -53,18 +53,9 @@ type Client struct {
 }
 
 func (tcpService *TcpService) genId() uint64 {
-	if node.GetNodeId()>MaxNodeId{
-		panic("nodeId exceeds the maximum!")
-	}
-
 	newSeed := atomic.AddUint32(&seed,1) % MaxSeed
 	nowTime := uint64(time.Now().Unix())%MaxTime
-	return (uint64(node.GetNodeId())<<50)|(nowTime<<19)|uint64(newSeed)
+	return (uint64(node.GetNodeId()%MaxNodeId)<<50)|(nowTime<<19)|uint64(newSeed)
-}
-
-
-func GetNodeId(agentId uint64) int {
-	return int(agentId>>50)
 }
 
 func (tcpService *TcpService) OnInit() error{

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 AddSafe[NumType NumberType](number1 NumType, number2 NumType) (NumType, bool) {
-func Add[NumType NumberType](number1 NumType, number2 NumType) NumType {
 	ret := number1 + number2
-	if number2> 0 && ret < number1 {
+	if number2 > 0 && ret < number1 {
-		log.Stack("Calculation overflow" ,log.Any("number1",number1),log.Any("number2",number2))
+		log.Stack("Calculation overflow", log.Any("number1", number1), log.Any("number2", number2))
-	}else if (number2<0 && ret > number1){
+		return ret, false
-		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))
+		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
+	ret, _ := SubSafe(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))
-	}
-
 	return ret
 }
 
-
 func Mul[NumType NumberType](number1 NumType, number2 NumType) NumType {
-	ret := number1 * number2
+	ret, _ := MulSafe(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))
 	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()))
+}
+