优化rpc与rankservice持久化

rpc/lclient.go

@@ -44,7 +44,8 @@ func (lc *LClient) Go(rpcHandler IRpcHandler,noReply bool, serviceMethod string,
 		sErr := errors.New("Call serviceMethod " + serviceMethod + " is error!")
 		log.SError(sErr.Error())
 		call := MakeCall()
-		call.Err = sErr
+		call.DoError(sErr)
+
 		return call
 	}
 
@@ -53,12 +54,13 @@ func (lc *LClient) Go(rpcHandler IRpcHandler,noReply bool, serviceMethod string,
 		//调用自己rpcHandler处理器
 		err := pLocalRpcServer.myselfRpcHandlerGo(lc.selfClient,serviceName, serviceMethod, args, requestHandlerNull,reply)
 		call := MakeCall()
+
 		if err != nil {
-			call.Err = err
+			call.DoError(err)
 			return call
 		}
 
-		call.done<-call
+		call.DoOK()
 		return call
 	}
 

rpc/rclient.go

@@ -44,8 +44,9 @@ func (rc *RClient) Go(rpcHandler IRpcHandler,noReply bool, serviceMethod string,
 	_, processor := GetProcessorType(args)
 	InParam, err := processor.Marshal(args)
 	if err != nil {
+		log.SError(err.Error())
 		call := MakeCall()
-		call.Err = err
+		call.DoError(err)
 		return call
 	}
 
@@ -65,14 +66,17 @@ func (rc *RClient) RawGo(rpcHandler IRpcHandler,processor IRpcProcessor, noReply
 
 	if err != nil {
 		call.Seq = 0
-		call.Err = err
+		log.SError(err.Error())
+		call.DoError(err)
 		return call
 	}
 
 	conn := rc.GetConn()
 	if conn == nil || conn.IsConnected()==false {
 		call.Seq = 0
-		call.Err = errors.New(serviceMethod + "  was called failed,rpc client is disconnect")
+		sErr := errors.New(serviceMethod + "  was called failed,rpc client is disconnect")
+		log.SError(sErr.Error())
+		call.DoError(sErr)
 		return call
 	}
 
@@ -83,8 +87,11 @@ func (rc *RClient) RawGo(rpcHandler IRpcHandler,processor IRpcProcessor, noReply
 	err = conn.WriteMsg([]byte{uint8(processor.GetProcessorType())}, bytes)
 	if err != nil {
 		rc.selfClient.RemovePending(call.Seq)
+
+		log.SError(err.Error())
+
 		call.Seq = 0
-		call.Err = err
+		call.DoError(err)
 	}
 
 	return call

rpc/rpc.go

@@ -102,6 +102,15 @@ func (rpcResponse *RpcResponse) Clear() *RpcResponse{
 	return rpcResponse
 }
 
+func (call *Call) DoError(err error){
+	call.Err = err
+	call.done <- call
+}
+
+func (call *Call) DoOK(){
+	call.done <- call
+}
+
 func (call *Call) Clear() *Call{
 	call.Seq = 0
 	call.ServiceMethod = ""

rpc/server.go

@@ -256,10 +256,10 @@ func (server *Server) selfNodeRpcHandlerGo(processor IRpcProcessor, client *Clie
 
 	rpcHandler := server.rpcHandleFinder.FindRpcHandler(handlerName)
 	if rpcHandler == nil {
+		err := errors.New("service method " + serviceMethod + " not config!")
+		log.SError(err.Error())
 		pCall.Seq = 0
-		pCall.Err = errors.New("service method " + serviceMethod + " not config!")
-		pCall.done <- pCall
-		log.SError(pCall.Err.Error())
+		pCall.DoError(err)
 
 		return pCall
 	}
@@ -273,10 +273,10 @@ func (server *Server) selfNodeRpcHandlerGo(processor IRpcProcessor, client *Clie
 		var err error
 		iParam,err = processor.Clone(args)
 		if err != nil {
+			sErr := errors.New("RpcHandler " + handlerName + "."+serviceMethod+" deep copy inParam is error:" + err.Error())
+			log.SError(sErr.Error())
 			pCall.Seq = 0
-			pCall.Err = errors.New("RpcHandler " + handlerName + "."+serviceMethod+" deep copy inParam is error:" + err.Error())
-			pCall.done <- pCall
-			log.SError(pCall.Err.Error())
+			pCall.DoError(sErr)
 
 			return pCall
 		}
@@ -289,9 +289,10 @@ func (server *Server) selfNodeRpcHandlerGo(processor IRpcProcessor, client *Clie
 		var err error
 		req.inParam, err = rpcHandler.UnmarshalInParam(processor, serviceMethod, rpcMethodId, rawArgs)
 		if err != nil {
+			log.SError(err.Error())
+			pCall.Seq = 0
+			pCall.DoError(err)
 			ReleaseRpcRequest(req)
-			pCall.Err = err
-			pCall.done <- pCall
 			return pCall
 		}
 	}
@@ -321,20 +322,22 @@ func (server *Server) selfNodeRpcHandlerGo(processor IRpcProcessor, client *Clie
 
 				return
 			}
+
 			if len(Err) == 0 {
 				v.Err = nil
+				v.DoOK()
 			} else {
-				v.Err = Err
+				log.SError(Err.Error())
+				v.DoError(Err)
 			}
-			v.done <- v
 		}
 	}
 
 	err := rpcHandler.PushRpcRequest(req)
 	if err != nil {
+		log.SError(err.Error())
+		pCall.DoError(err)
 		ReleaseRpcRequest(req)
-		pCall.Err = err
-		pCall.done <- pCall
 	}
 
 	return pCall

sysservice/rankservice/MongodbPersist.go

@@ -6,9 +6,9 @@ import (
 	"github.com/duanhf2012/origin/rpc"
 	"github.com/duanhf2012/origin/service"
 	"github.com/duanhf2012/origin/sysmodule/mongodbmodule"
-	"github.com/duanhf2012/origin/util/coroutine"
 	"go.mongodb.org/mongo-driver/bson"
 	"go.mongodb.org/mongo-driver/mongo/options"
+	"runtime"
 	"sync"
 	"sync/atomic"
 	"time"
@@ -71,7 +71,8 @@ func (mp *MongoPersist) OnInit() error {
 	}
 
 	//开启协程
-	coroutine.GoRecover(mp.persistCoroutine,-1)
+	go mp.persistCoroutine()
+
 	return nil
 }
 
@@ -291,6 +292,15 @@ func (mp *MongoPersist) hasPersistData() bool{
 }
 
 func (mp *MongoPersist) saveToDB(){
+	defer func() {
+		if r := recover(); r != nil {
+			buf := make([]byte, 4096)
+			l := runtime.Stack(buf, false)
+			errString := fmt.Sprint(r)
+			log.SError(" Core dump info[", errString, "]\n", string(buf[:l]))
+		}
+	}()
+
 	//1.copy数据
 	mp.Lock()
 	mapRemoveRankData := mp.mapRemoveRankData