Golang原生rpc（rpc服务端源码解读）

落日映苍穹つ 2022-11-13 11:28 223阅读 0赞

创建rpc接口，需要几个条件

*  方法的类型是可输出的
 *  方法的本身也是可输出的
 *  方法必须有两个参数，必须是输出类型或者是内建类型
 *  方法的第二个参数是指针类型
 *  方法返回的类型为error

## rpc服务原理分析 ##

**server端**

*  服务注册
 *  处理网络调用  
    *服务注册*  
    通过反射处理，将接口存入到map中，进行调用  
    注册服务两个方法

func Register (rcvr interface{
        }) error {
        }
    func RegisterName (rcvr interface{
        } , name string) error {
        }
    //指定注册的名称

**注册方法的源代码解读**  
首先，无论是Register还是RegisterName底层代码都是调用register方法，进行服务注册。  
server.go  
**register方法解读**

func (server *Server) register(rcvr interface{
        }, name string, useName bool) error {
        
    	//创建一个service实例
    	s := new(service)
    	s.typ = reflect.TypeOf(rcvr)
    	s.rcvr = reflect.ValueOf(rcvr)
    	sname := reflect.Indirect(s.rcvr).Type().Name()
    	//如果服务名为空，则使用默认的服务名
    	if useName {
        
    		sname = name
    	}
    	if sname == "" {
        
    		s := "rpc.Register: no service name for type " + s.typ.String()
    		log.Print(s)
    		return errors.New(s)
    	}
    	//判断方法名是否暴漏的，如果方法名不是暴露的，则会导致调用不成功，所以返回false
    	if !token.IsExported(sname) && !useName {
        
    		s := "rpc.Register: type " + sname + " is not exported"
    		log.Print(s)
    		return errors.New(s)
    	}
    	s.name = sname
    
    	// Install the methods
    	//调用suitableMethods函数，进行返回接口，在suitableMethods中判断方法是否符合作为rpc接口的条件，如果符合，则进行添加到services中
    	s.method = suitableMethods(s.typ, true)
    
    	if len(s.method) == 0 {
        
    		str := ""
    
    		// To help the user, see if a pointer receiver would work.
    		//如果方法绑定到结构体的地址上，使用reflect.TypeOf()是不会发现方法的，所以也要进行查找绑定到结构体地址上的方法
    		method := suitableMethods(reflect.PtrTo(s.typ), false)
    		if len(method) != 0 {
        
    			str = "rpc.Register: type " + sname + " has no exported methods of suitable type (hint: pass a pointer to value of that type)"
    		} else {
        
    			str = "rpc.Register: type " + sname + " has no exported methods of suitable type"
    		}
    		log.Print(str)
    		return errors.New(str)
    	}
    	//判断服务接口是否已经注册。
    	if _, dup := server.serviceMap.LoadOrStore(sname, s); dup {
        
    		return errors.New("rpc: service already defined: " + sname)
    	}
    	return nil
    }

**suitableMethod方法解读**

func suitableMethods(typ reflect.Type, reportErr bool) map[string]*methodType {
        
    	//创建一个方法的切片
    	methods := make(map[string]*methodType)
    	for m := 0; m < typ.NumMethod(); m++ {
        
    		method := typ.Method(m)
    		mtype := method.Type
    		mname := method.Name
    		// Method must be exported.
    		if method.PkgPath != "" {
        
    			continue
    		}
    		// Method needs three ins: receiver, *args, *reply.
    		//如果传入的参数，不为三个，则会报错，这里为什么是三个？
    		//golang方法体中默认传入结构体实例，所以request,*response,结构体实例一共三个参数
    		if mtype.NumIn() != 3 {
        
    			if reportErr {
        
    				log.Printf("rpc.Register: method %q has %d input parameters; needs exactly three\n", mname, mtype.NumIn())
    			}
    			continue
    		}
    		// First arg need not be a pointer.
    		argType := mtype.In(1)
    		if !isExportedOrBuiltinType(argType) {
        
    			if reportErr {
        
    				log.Printf("rpc.Register: argument type of method %q is not exported: %q\n", mname, argType)
    			}
    			continue
    		}
    		// Second arg must be a pointer.
    		//判断第二个参数是否为指针，如果不为指针，则返回false。
    		replyType := mtype.In(2)
    		if replyType.Kind() != reflect.Ptr {
        
    			if reportErr {
        
    				log.Printf("rpc.Register: reply type of method %q is not a pointer: %q\n", mname, replyType)
    			}
    			continue
    		}
    		// Reply type must be exported.
    		if !isExportedOrBuiltinType(replyType) {
        
    			if reportErr {
        
    				log.Printf("rpc.Register: reply type of method %q is not exported: %q\n", mname, replyType)
    			}
    			continue
    		}
    		// Method needs one out.
    		//返回结果是否为一个值，且为error
    		if mtype.NumOut() != 1 {
        
    			if reportErr {
        
    				log.Printf("rpc.Register: method %q has %d output parameters; needs exactly one\n", mname, mtype.NumOut())
    			}
    			continue
    		}
    		// The return type of the method must be error.
    		if returnType := mtype.Out(0); returnType != typeOfError {
        
    			if reportErr {
        
    				log.Printf("rpc.Register: return type of method %q is %q, must be error\n", mname, returnType)
    			}
    			continue
    		}
    		//将接口加入service
    		methods[mname] = &methodType{
        method: method, ArgType: argType, ReplyType: replyType}
    	}
    	return methods
    }

接收到请求后会不断的解析请求  
解析请求的两个方法  
readRequestHeader

func (server *Server) readRequestHeader(codec ServerCodec) (svc *service, mtype *methodType, req *Request, keepReading bool, err error) {
        
    	// Grab the request header.
    	//接收到请求，对请求进行编码
    	req = server.getRequest()
    	err = codec.ReadRequestHeader(req)
    	if err != nil {
        
    		req = nil
    		if err == io.EOF || err == io.ErrUnexpectedEOF {
        
    			return
    		}
    		err = errors.New("rpc: server cannot decode request: " + err.Error())
    		return
    	}
    
    	// We read the header successfully. If we see an error now,
    	// we can still recover and move on to the next request.
    	keepReading = true
    //编码后的请求，进行间隔，所以只要进行将.的左右两边的数据进行分割，就能解码
    	dot := strings.LastIndex(req.ServiceMethod, ".")
    	if dot < 0 {
        
    		err = errors.New("rpc: service/method request ill-formed: " + req.ServiceMethod)
    		return
    	}
    	serviceName := req.ServiceMethod[:dot]
    	methodName := req.ServiceMethod[dot+1:]
    
    	// Look up the request.
    	svci, ok := server.serviceMap.Load(serviceName)
    	if !ok {
        
    		err = errors.New("rpc: can't find service " + req.ServiceMethod)
    		return
    	}
    	svc = svci.(*service)
    	//获取到注册服务时，注册的接口
    	mtype = svc.method[methodName]
    	if mtype == nil {
        
    		err = errors.New("rpc: can't find method " + req.ServiceMethod)
    	}
    	return
    }

readRequest方法

func (server *Server) readRequest(codec ServerCodec) (service *service, mtype *methodType, req *Request, argv, replyv reflect.Value, keepReading bool, err error) {
        
    	service, mtype, req, keepReading, err = server.readRequestHeader(codec)
    //调用上面的readRequestHeader方法，进行解码，并返返回接口数据
    	if err != nil {
        
    		if !keepReading {
        
    			return
    		}
    		// discard body
    		codec.ReadRequestBody(nil)
    		return
    	}
    
    	// Decode the argument value.
    	argIsValue := false // if true, need to indirect before calling.
    	//判断传擦是否为指针，如果为指针，需要使用Elem()方法，进行指向结构体
    	if mtype.ArgType.Kind() == reflect.Ptr {
        
    		argv = reflect.New(mtype.ArgType.Elem())
    	} else {
        
    		argv = reflect.New(mtype.ArgType)
    		argIsValue = true
    	}
    	// argv guaranteed to be a pointer now.
    	if err = codec.ReadRequestBody(argv.Interface()); err != nil {
        
    		return
    	}
    	if argIsValue {
        
    		argv = argv.Elem()
    	}
    
    	replyv = reflect.New(mtype.ReplyType.Elem())
    
    	switch mtype.ReplyType.Elem().Kind() {
        
    	case reflect.Map:
    		replyv.Elem().Set(reflect.MakeMap(mtype.ReplyType.Elem()))
    	case reflect.Slice:
    		replyv.Elem().Set(reflect.MakeSlice(mtype.ReplyType.Elem(), 0, 0))
    	}
    	return
    }

call方法

func (s *service) call(server *Server, sending *sync.Mutex, wg *sync.WaitGroup, mtype *methodType, req *Request, argv, replyv reflect.Value, codec ServerCodec) {
        
    	if wg != nil {
        
    		defer wg.Done()
    	}
    	mtype.Lock()
    	mtype.numCalls++
    	mtype.Unlock()
    	function := mtype.method.Func
    	// Invoke the method, providing a new value for the reply.
    	//调用call方法，并将参数转化为valueof型参数，
    	returnValues := function.Call([]reflect.Value{
        s.rcvr, argv, replyv})
    	// The return value for the method is an error.
    	//将返回的error进行读取，转化为interface{}型
    	errInter := returnValues[0].Interface()
    	errmsg := ""
    	if errInter != nil {
        
    	//将error进行断言
    		errmsg = errInter.(error).Error()
    	}
    	server.sendResponse(sending, req, replyv.Interface(), codec, errmsg)
    	server.freeRequest(req)
    }

**注册的大概流程**

1.  根据反射，进行接口的获取
2.  使用方法判断接口是否符合作为rpc接口的规范（有两个参数，第二个参数为指针，返回一个参数error）
3.  如果不符合规范，将返回error，符合规范，将存入map，进行提供调用

**接收请求的大概流程**

1.  首先，不断的接收数据流，并进行解码，解码之后为data.data，所以我们需要使用 . 作为分隔符，进行数据的截切和读取
2.  将读取的数据在注册的map中进行查找，如果查找到，返回相关的service和其他数据
3.  进行调用