From 7e6b3098d48ff523aff2a75a452d283d645f2555 Mon Sep 17 00:00:00 2001 From: duanhf2012 Date: Wed, 1 Apr 2020 14:22:31 +0800 Subject: [PATCH] =?UTF-8?q?=E5=8E=BB=E6=8E=89GR=E6=A8=A1=E5=BC=8F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- rpc/client.go | 6 +--- rpc/rpchandler.go | 28 ++++------------ rpc/server.go | 82 +++++++++++++++++++++++------------------------ 3 files changed, 47 insertions(+), 69 deletions(-) diff --git a/rpc/client.go b/rpc/client.go index 862c220..5237959 100644 --- a/rpc/client.go +++ b/rpc/client.go @@ -62,7 +62,6 @@ func (slf *Client) AsycGo(rpcHandler IRpcHandler,mutiCoroutine bool,serviceMetho request := &RpcRequest{} request.NoReply = false - request.MutiCoroutine = mutiCoroutine call.Arg = args slf.pendingLock.Lock() slf.startSeq += 1 @@ -92,13 +91,12 @@ func (slf *Client) AsycGo(rpcHandler IRpcHandler,mutiCoroutine bool,serviceMetho return call.Err } -func (slf *Client) Go(noReply bool,mutiCoroutine bool,serviceMethod string, args interface{},reply interface{}) *Call { +func (slf *Client) Go(noReply bool,serviceMethod string, args interface{},reply interface{}) *Call { call := new(Call) call.done = make(chan *Call,1) call.Reply = reply request := &RpcRequest{} - request.MutiCoroutine = mutiCoroutine request.NoReply = noReply call.Arg = args slf.pendingLock.Lock() @@ -137,7 +135,6 @@ type RpcRequest struct { Seq uint64 // sequence number chosen by client ServiceMethod string // format: "Service.Method" NoReply bool //是否需要返回 - MutiCoroutine bool // 是否多协程模式 //packbody InParam []byte @@ -193,7 +190,6 @@ func (slf *Client) Run(){ v.Err= respone.Err } - if v.callback.IsValid() { v.rpcHandler.(*RpcHandler).callResponeCallBack<-v }else{ diff --git a/rpc/rpchandler.go b/rpc/rpchandler.go index 4f77825..40c377a 100644 --- a/rpc/rpchandler.go +++ b/rpc/rpchandler.go @@ -64,11 +64,8 @@ type IRpcHandler interface { CallMethod(ServiceMethod string,param interface{},reply interface{}) error AsyncCall(serviceMethod string,args interface{},callback interface{}) error - GRAsyncCall(serviceMethod string,args interface{},callback interface{}) error Call(serviceMethod string,args interface{},reply interface{}) error - GRCall(serviceMethod string,args interface{},reply interface{}) error Go(serviceMethod string,args interface{}) error - GRGo(serviceMethod string,args interface{}) error AsyncCallNode(nodeId int,serviceMethod string,args interface{},callback interface{}) error CallNode(nodeId int,serviceMethod string,args interface{},reply interface{}) error GoNode(nodeId int,serviceMethod string,args interface{}) error @@ -282,7 +279,7 @@ func (slf *RpcHandler) goRpc(bCast bool,nodeId int,serviceMethod string,mutiCoro return pLocalRpcServer.myselfRpcHandlerGo(sMethod[0],sMethod[1],args,nil) } //其他的rpcHandler的处理器 - pCall := pLocalRpcServer.rpcHandlerGo(true,mutiCoroutine,sMethod[0],sMethod[1],args,nil) + pCall := pLocalRpcServer.rpcHandlerGo(true,sMethod[0],sMethod[1],args,nil) if pCall.Err!=nil { err = pCall.Err } @@ -290,7 +287,7 @@ func (slf *RpcHandler) goRpc(bCast bool,nodeId int,serviceMethod string,mutiCoro } //跨node调用 - pCall := pClient.Go(true,mutiCoroutine,serviceMethod,args,nil) + pCall := pClient.Go(true,serviceMethod,args,nil) if pCall.Err!=nil { err = pCall.Err } @@ -329,13 +326,13 @@ func (slf *RpcHandler) callRpc(nodeId int,serviceMethod string,mutiCoroutine boo return pLocalRpcServer.myselfRpcHandlerGo(sMethod[0],sMethod[1],args,reply) } //其他的rpcHandler的处理器 - pCall := pLocalRpcServer.rpcHandlerGo(false,mutiCoroutine,sMethod[0],sMethod[1],args,reply) + pCall := pLocalRpcServer.rpcHandlerGo(false,sMethod[0],sMethod[1],args,reply) pResult := pCall.Done() return pResult.Err } //跨node调用 - pCall := pClient.Go(false,mutiCoroutine,serviceMethod,args,reply) + pCall := pClient.Go(false,serviceMethod,args,reply) pResult := pCall.Done() return pResult.Err } @@ -387,13 +384,13 @@ func (slf *RpcHandler) asyncCallRpc(nodeid int,serviceMethod string,mutiCoroutin //其他的rpcHandler的处理器 if callback!=nil { - err = pLocalRpcServer.rpcHandlerAsyncGo(slf,false,mutiCoroutine,sMethod[0],sMethod[1],args,reply,fVal) + err = pLocalRpcServer.rpcHandlerAsyncGo(slf,false,sMethod[0],sMethod[1],args,reply,fVal) if err != nil { fVal.Call([]reflect.Value{reflect.ValueOf(reply),reflect.ValueOf(err)}) } return nil } - pCall := pLocalRpcServer.rpcHandlerGo(false,mutiCoroutine,sMethod[0],sMethod[1],args,reply) + pCall := pLocalRpcServer.rpcHandlerGo(false,sMethod[0],sMethod[1],args,reply) pResult := pCall.Done() return pResult.Err } @@ -419,32 +416,19 @@ func (slf *RpcHandler) AsyncCall(serviceMethod string,args interface{},callback return slf.asyncCallRpc(0,serviceMethod,false,args,callback) } -func (slf *RpcHandler) GRAsyncCall(serviceMethod string,args interface{},callback interface{}) error { - return slf.asyncCallRpc(0,serviceMethod,true,args,callback) -} - func (slf *RpcHandler) Call(serviceMethod string,args interface{},reply interface{}) error { return slf.callRpc(0,serviceMethod,false,args,reply) } -func (slf *RpcHandler) GRCall(serviceMethod string,args interface{},reply interface{}) error { - return slf.callRpc(0,serviceMethod,true,args,reply) -} func (slf *RpcHandler) Go(serviceMethod string,args interface{}) error { return slf.goRpc(false,0,serviceMethod,false,args) } -func (slf *RpcHandler) GRGo(serviceMethod string,args interface{}) error { - return slf.goRpc(false,0,serviceMethod,true,args) -} - - func (slf *RpcHandler) AsyncCallNode(nodeId int,serviceMethod string,args interface{},callback interface{}) error { return slf.asyncCallRpc(nodeId,serviceMethod,false,args,callback) } - func (slf *RpcHandler) CallNode(nodeId int,serviceMethod string,args interface{},reply interface{}) error { return slf.callRpc(nodeId,serviceMethod,false,args,reply) } diff --git a/rpc/server.go b/rpc/server.go index 828049b..6709fb7 100644 --- a/rpc/server.go +++ b/rpc/server.go @@ -86,6 +86,31 @@ type RpcRequestRw struct { requestHandle RequestHandler } +func (agent *RpcAgent) WriteRespone(serviceMethod string,seq uint64,reply interface{},err *RpcError) { + var rpcRespone RpcResponse + rpcRespone.Seq = seq + rpcRespone.Err = err + var errM error + if reply!=nil { + rpcRespone.Returns,errM = processor.Marshal(reply) + if errM!= nil { + rpcRespone.Err = ConvertError(errM) + } + } + + bytes,errM := processor.Marshal(&rpcRespone) + if errM != nil { + log.Error("service method %s %+v Marshal error:%+v!", serviceMethod,rpcRespone,errM) + return + } + + errM = agent.conn.WriteMsg(bytes) + if errM != nil { + log.Error("Rpc %s return is error:%+v",serviceMethod,errM) + } +} + + func (agent *RpcAgent) Run() { for { data,err := agent.conn.ReadMsg() @@ -111,46 +136,26 @@ func (agent *RpcAgent) Run() { //交给程序处理 serviceMethod := strings.Split(req.ServiceMethod,".") if len(serviceMethod)!=2 { + rpcError := RpcError("rpc request req.ServiceMethod is error") + agent.WriteRespone(req.ServiceMethod,req.Seq,nil,&rpcError) log.Debug("rpc request req.ServiceMethod is error") continue } rpcHandler := agent.rpcserver.rpcHandleFinder.FindRpcHandler(serviceMethod[0]) if rpcHandler== nil { + rpcError := RpcError(fmt.Sprintf("service method %s not config!", req.ServiceMethod)) + agent.WriteRespone(req.ServiceMethod,req.Seq,nil,&rpcError) log.Error("service method %s not config!", req.ServiceMethod) continue } + if req.NoReply == false { req.requestHandle = func(Returns interface{},Err *RpcError){ - var rpcRespone RpcResponse - rpcRespone.Seq = req.Seq - rpcRespone.Err = Err - if Err==nil { - rpcRespone.Returns,err = processor.Marshal(Returns) - if err!= nil { - rpcRespone.Err = ConvertError(err) - } - //rpcRespone.Returns, = processor.Marshal(Returns) - } - - bytes,err := processor.Marshal(&rpcRespone) - if err != nil { - log.Error("service method %s Marshal error:%+v!", req.ServiceMethod,err) - return - } - - err = agent.conn.WriteMsg(bytes) - if err != nil { - log.Error("Rpc %s return is error:%+v",req.ServiceMethod,err) - } + agent.WriteRespone(req.ServiceMethod,req.Seq,Returns,Err) } } - if req.MutiCoroutine == true { - go rpcHandler.HandlerRpcRequest(&req) - }else{ - rpcHandler.PushRequest(&req) - } - + rpcHandler.PushRequest(&req) } } @@ -195,7 +200,7 @@ func (slf *Server) myselfRpcHandlerGo(handlerName string,methodName string, args } -func (slf *Server) rpcHandlerGo(noReply bool,mutiCoroutine bool,handlerName string,methodName string, args interface{},reply interface{}) *Call { +func (slf *Server) rpcHandlerGo(noReply bool,handlerName string,methodName string, args interface{},reply interface{}) *Call { pCall := &Call{} pCall.done = make( chan *Call,1) rpcHandler := slf.rpcHandleFinder.FindRpcHandler(handlerName) @@ -218,16 +223,13 @@ func (slf *Server) rpcHandlerGo(noReply bool,mutiCoroutine bool,handlerName stri } } - if mutiCoroutine == true { - go rpcHandler.HandlerRpcRequest(&req) - }else{ - rpcHandler.PushRequest(&req) - } + + rpcHandler.PushRequest(&req) return pCall } -func (slf *Server) rpcHandlerAsyncGo(callerRpcHandler IRpcHandler,noReply bool,mutiCoroutine bool,handlerName string,methodName string,args interface{},reply interface{},callback reflect.Value) error { +func (slf *Server) rpcHandlerAsyncGo(callerRpcHandler IRpcHandler,noReply bool,handlerName string,methodName string,args interface{},reply interface{},callback reflect.Value) error { pCall := &Call{} //pCall.done = make( chan *Call,1) pCall.rpcHandler = callerRpcHandler @@ -245,7 +247,6 @@ func (slf *Server) rpcHandlerAsyncGo(callerRpcHandler IRpcHandler,noReply bool,m req.localParam = args req.localReply = reply req.NoReply = noReply - req.MutiCoroutine = mutiCoroutine if noReply == false { req.requestHandle = func(Returns interface{},Err *RpcError){ @@ -257,13 +258,10 @@ func (slf *Server) rpcHandlerAsyncGo(callerRpcHandler IRpcHandler,noReply bool,m } } - if mutiCoroutine == true { - go rpcHandler.HandlerRpcRequest(&req) - }else{ - err := rpcHandler.PushRequest(&req) - if err != nil { - return err - } + + err := rpcHandler.PushRequest(&req) + if err != nil { + return err } return nil