Golang 1.4 net/rpc server源码介绍
发布时间:2021-11-19 14:45:33 所属栏目:教程 来源:互联网
导读:方法注册: 因为从client我们知道是复用的socket来实现并发调用rpc方法,我们先从方法注册来看源码部分: // Server对象大都是保存方法存根,保证对象互斥的 type Serverstruct{ mu sync.RWMutex// protects the serviceMap serviceMap map[string]*service
方法注册: 因为从client我们知道是复用的socket来实现并发调用rpc方法,我们先从方法注册来看源码部分: // Server对象大都是保存方法存根,保证对象互斥的 type Serverstruct{ mu sync.RWMutex// protects the serviceMap serviceMap map[string]*service reqLock sync.Mutex// protects freeReq freeReq *Request respLock sync.Mutex// protects freeResp freeResp *Response} func NewServer()*Server{return&Server{serviceMap: make(map[string]*service)}}// rpc.Register默认使用了一个Server,只对serviceMap进行了初始化 varDefaultServer=NewServer()// rpc的service包括方法名、方法反射,类型等 type service struct{ name string// name of service rcvr reflect.Value// receiver of methods for the service typ reflect.Type// type of the receiver method map[string]*methodType // registered methods }// 无论是RegisterName、Register最终都调用了register的内部方法 func (server *Server)register(rcvr interface{}, name string, useName bool) error {// 保证注册服务安全,先加锁 server.mu.Lock() defer server.mu.Unlock()// 如果服务为空,默认注册一个 if server.serviceMap ==nil{ server.serviceMap = make(map[string]*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)}// 方法必须是暴露的,既服务名首字符大写 if!isExported(sname)&&!useName { s :="rpc.Register: type "+ sname +" is not exported" log.Print(s)return errors.New(s)}// 不允许重复注册 if _, present := server.serviceMap[sname]; present {return errors.New("rpc: service already defined: "+ sname)} s.name = sname // 开始注册rpc struct内部的方法存根 s.method = suitableMethods(s.typ,true)// 如果struct内部一个方法也没,那么直接报错,错误信息还非常详细 if len(s.method)==0{ str :=""// To help the user, see if a pointer receiver would work. 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)}// 保存在server的serviceMap中 server.serviceMap[s.name]= s returnnil}// 上文提到了服务还需要方法存根的注册 func suitableMethods(typ reflect.Type, reportErr bool) map[string]*methodType {// 根据方法名创建保存内部方法map methods := make(map[string]*methodType)// 获取rpc struct内部的方法 for m :=0; m < typ.NumMethod(); m++{ method := typ.Method(m) mtype := method.Type mname := method.Name// 之前对这行代码觉得比较奇葩,方法是否是暴露,是看是否有PkgPath的,如果是私有方法,PkgPath显示包名 if method.PkgPath!=""{continue}// 判断是否是三个参数:第一个是结构本身,第二个是参数,第三个是返回值 // Method needs three ins: receiver, *args, *reply. if mtype.NumIn()!=3{if reportErr { log.Println("method", mname,"has wrong number of ins:", mtype.NumIn())}continue}// args是指针类型 // First arg need not be a pointer. argType := mtype.In(1)if!isExportedOrBuiltinType(argType){if reportErr { log.Println(mname,"argument type not exported:", argType)}continue}// reply是指针类型 // Second arg must be a pointer. replyType := mtype.In(2)if replyType.Kind()!= reflect.Ptr{if reportErr { log.Println("method", mname,"reply type not a pointer:", replyType)}continue}// Reply type must be exported. // reply必须是可暴露的 if!isExportedOrBuiltinType(replyType){if reportErr { log.Println("method", mname,"reply type not exported:", replyType)}continue}// Method needs one out. // 必须有一个返回值,而且要是error if mtype.NumOut()!=1{if reportErr { log.Println("method", mname,"has wrong number of outs:", mtype.NumOut())}continue}// The return type of the method must be error. if returnType := mtype.Out(0); returnType != typeOfError {if reportErr { log.Println("method", mname,"returns", returnType.String(),"not error")}continue} methods[mname]=&methodType{method: method,ArgType: argType,ReplyType: replyType}}return methods } 请求调用: 方法已经被注册成功,接下来我们看看是如何客户端发送请求调用的: func (server *Server)Accept(lis net.Listener){for{ conn, err := lis.Accept()if err !=nil{ log.Fatal("rpc.Serve: accept:", err.Error())// TODO(r): exit? }// accept连接以后,打开一个goroutine处理请求 go server.ServeConn(conn)}} func (server *Server)ServeConn(conn io.ReadWriteCloser){ buf := bufio.NewWriter(conn) srv :=&gobServerCodec{ rwc: conn, dec: gob.NewDecoder(conn), enc: gob.NewEncoder(buf), encBuf: buf,}// 根据指定的codec进行协议解析 server.ServeCodec(srv)} func (server *Server)ServeCodec(codec ServerCodec){ sending :=new(sync.Mutex)for{// 解析请求 service, mtype, req, argv, replyv, keepReading, err := server.readRequest(codec)if err !=nil{if debugLog && err != io.EOF { log.Println("rpc:", err)}if!keepReading {break}// send a response if we actually managed to read a header. // 如果当前请求错误了,我们应该返回信息,然后继续处理 if req !=nil{ server.sendResponse(sending, req, invalidRequest, codec, err.Error()) server.freeRequest(req)}continue}// 因为需要继续处理后续请求,所以开一个gorutine处理rpc方法 go service.call(server, sending, mtype, req, argv, replyv, codec)}// 如果连接关闭了需要释放资源 codec.Close()} func (server *Server) readRequestHeader(codec ServerCodec)(service *service, mtype *methodType, req *Request, keepReading bool, err error){// 解析头部,如果失败,直接返回了 req = server.getRequest() err = codec.ReadRequestHeader(req)if err !=nil{ req =nilif err == io.EOF || err == io.ErrUnexpectedEOF{return} err = errors.New("rpc: server cannot decode request: "+ err.Error())return}if debugLog { log.Printf("rpc: [trace:%v]n", req.Tracer)}// We read the header successfully. If we see an error now, // we can still recover and move on to the next request. keepReading =true// 获取请求中xxx.xxx中.的位置 dot := strings.LastIndex(req.ServiceMethod,".")if dot <0{ err = errors.New("rpc: service/method request ill-formed: "+ req.ServiceMethod)return}// 拿到struct名字和方法名字 serviceName := req.ServiceMethod[:dot] methodName := req.ServiceMethod[dot+1:]// Look up the request.// 加读锁,获取对象 server.mu.RLock() service = server.serviceMap[serviceName] server.mu.RUnlock()if service ==nil{ err = errors.New("rpc: can't find service "+ req.ServiceMethod)return}// 获取反射类型,看见rpc中的发射其实是预先放入map中的 mtype = service.method[methodName]if mtype ==nil{ err = errors.New("rpc: can't find method "+ req.ServiceMethod)}return} 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)if err !=nil{if!keepReading {return}// discard body codec.ReadRequestBody(nil)return}// 解析请求中的args argIsValue :=false// if true, need to indirect before calling. 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()}// 初始化reply类型 replyv = reflect.New(mtype.ReplyType.Elem())return} func (s *service) call(server *Server, sending *sync.Mutex, mtype *methodType, req *Request, argv, replyv reflect.Value, codec ServerCodec){ mtype.Lock() mtype.numCalls++ mtype.Unlock()function:= mtype.method.Func// Invoke the method, providing a new value for the reply. // 这里是真正调用rpc方法的地方 returnValues :=function.Call([]reflect.Value{s.rcvr, argv, replyv})// The return value for the method is an error. errInter := returnValues[0].Interface() errmsg :=""if errInter !=nil{ errmsg = errInter.(error).Error()}// 处理返回请求了 server.sendResponse(sending, req, replyv.Interface(), codec, errmsg) server.freeRequest(req)} func (server *Server) sendResponse(sending *sync.Mutex, req *Request, reply interface{}, codec ServerCodec, errmsg string){ resp := server.getResponse()// Encode the response header resp.ServiceMethod= req.ServiceMethodif errmsg !=""{ resp.Error= errmsg reply = invalidRequest }// 上一文提到,客户端是根据序号来定位请求的,所以需要原样返回 resp.Seq= req.Seq sending.Lock() err := codec.WriteResponse(resp, reply)if debugLog && err !=nil{ log.Println("rpc: writing response:", err)} sending.Unlock() server.freeResponse(resp)} 资源重用: 上面把大致的rpc请求都说明了,server有一个技巧是重用对象,这里使用的是链表方式处理的: // 可以看出使用一个free list链表,来避免Request以及Response对象频繁创建,导致GC压力 func (server *Server) getRequest()*Request{ server.reqLock.Lock() req := server.freeReq if req ==nil{ req =new(Request)}else{ server.freeReq = req.next*req =Request{}} server.reqLock.Unlock()return req } func (server *Server) freeRequest(req *Request){ server.reqLock.Lock() req.next= server.freeReq server.freeReq = req server.reqLock.Unlock()} func (server *Server) getResponse()*Response{ server.respLock.Lock() resp := server.freeResp if resp ==nil{ resp =new(Response)}else{ server.freeResp = resp.next*resp =Response{}} server.respLock.Unlock()return resp } func (server *Server) freeResponse(resp *Response){ server.respLock.Lock() resp.next= server.freeResp server.freeResp = resp server.respLock.Unlock()} 最后,sending这把锁的目的是避免同一个套接字快速请求中避免返回包写入乱序,因此避免一个包完整写入完毕才允许下一个返回写入套接字。通过rpc包源码解析,可以看到标准库中的核心思想还是channel+mutex实现复用对象,以及各种方式的复用,避免GC压力,在我们以后写高性能服务端可以借鉴的地方。 (编辑:济南站长网) 【声明】本站内容均来自网络,其相关言论仅代表作者个人观点,不代表本站立场。若无意侵犯到您的权利,请及时与联系站长删除相关内容! |