Golang原生rpc
鹿灏楷silves 人气: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) }
注册的大概流程
- 根据反射,进行接口的获取
- 使用方法判断接口是否符合作为rpc接口的规范(有两个参数,第二个参数为指针,返回一个参数error)
- 如果不符合规范,将返回error,符合规范,将存入map,进行提供调用
接收请求的大概流程
- 首先,不断的接收数据流,并进行解码,解码之后为data.data,所以我们需要使用 . 作为分隔符,进行数据的截切和读取
- 将读取的数据在注册的map中进行查找,如果查找到,返回相关的service和其他数据
- 进行调用
加载全部内容