【GeeRPC】Day3：服务注册（Service Register）

Day3：服务注册（Service Register）

今天的任务是：

• 通过反射实现服务注册功能；
• 在服务端实现服务调用，代码约 150 行；

结构体映射为服务

RPC 框架的一个基本能力是：像调用本地程序一样调用远程服务。如何将程序映射为服务呢？对于 Golang 而言，这个问题就变成了如何将结构体的方法映射为服务。

对 net/rpc 而言，一个函数需要能够被远程调用，需要满足如下五个条件：

• the method’s type is exported. 方法所属的类型是导出的；
• the method’s is exported. 方法是导出的；
• the method has two arguments, both exported (or builtin) types. 两个入参，均为导出或内置类型；
• this method’s second argument is a pointer. 第二个入参必须是指针；
• this method has return type error. 返回值为 error 类型。

更直观一些：

func (t *T) MethodName(argType T1, replyType *T2) error

假设客户端发送过来一个请求，包含 ServiceMethod 和 Argv：

{"ServiceMethod": "T.MethodName""Argv": "0101110101..."			// 序列化之后的字节流
}

通过"T.MethodName"可以确定调用的是类型 T 的 MethodName，如果硬编码实现这个功能，可能是这样的：

switch req.ServiceMethod {case "T.MethodName":t := new(T)reply := new(T2)var argv T1gob.NewDecoder(conn).Decode(&argv)err := t.MethodName(argv, reply)server.sendMessage(reply, err)case "Foo.Sum":f := new(Foo)...
}

即，如果采用硬编码的方式来实现结构体与服务的映射，每暴露一个方法，就需要编写等量的代码。借助反射，我们才能映射过程的自动化。

通过反射，我们可以非常容易地获取某个结构体的所有方法，并且能够通过方法，获得到该方法所有的参数类型和返回值：

func main() {var wg sync.WaitGrouptyp := reflect.TypeOf(&wg)for i := 0; i < typ.NumMethod(); i++ {method := typ.Method(i)argv := make([]string, 0, method.Type.NumIn())returns := make([]string, 0, method.Type.NumOut())// j 从 1 开始，第 0 个入参是 wg 自己。for j := 1; j < method.Type.NumIn(); j++ {argv = append(argv, method.Type.In(j).Name())}for j := 0; j < method.Type.NumOut(); j++ {returns = append(returns, method.Type.Out(j).Name())}log.Printf("func (w *%s) %s(%s) %s",typ.Elem().Name(),method.Name,strings.Join(argv, ","),strings.Join(returns, ","))}
}// OUTPUT:
func (w *WaitGroup) Add(int)
func (w *WaitGroup) Done()
func (w *WaitGroup) Wait()

通过反射实现 service

目前我们服务端的功能不是很完整，仅仅将请求的 header 打印了出来，而没有真正地处理。今天的目的是补全这部分功能。首先我们通过反射实现结构体与服务的映射关系，代码独立放置在service.go当中。

package geerpcimport ("reflect""sync/atomic"
)// in service/service.gotype methodType struct {method    reflect.Method	// 反射方法对象，包含方法的信息ArgType   reflect.Type		// 方法的参数类型ReplyType reflect.Type		// 方法的返回值类型numCalls  uint64			// 方法被调用的次数
}func (m *methodType) NumCalls() uint64 {return atomic.LoadUint64(&m.numCalls)
}func (m *methodType) newArgv() reflect.Value {var argv reflect.Value// arg may be a pointer type or a value typeif m.ArgType.Kind() == reflect.Ptr {argv = reflect.New(m.ArgType.Elem())} else {argv = reflect.New(m.ArgType).Elem()}return argv		// 返回一个 reflect.Value 对象, 表示新创建的参数值.
}func (m *methodType) newReplyv() reflect.Value {// reply must be a pointer typereplyv := reflect.New(m.ReplyType.Elem())switch m.ReplyType.Elem().Kind() {case reflect.Map:replyv.Elem().Set(reflect.MakeMap(m.ReplyType.Elem()))case reflect.Slice:replyv.Elem().Set(reflect.MakeSlice(m.ReplyType.Elem(), 0, 0))}return replyv		// 返回一个 reflect.Value 对象, 表示新创建的返回值.
}

每一个 methodType 实例包含了一个方法的完整信息。包括：

• method：方法本身；
• ArgType：第一个参数的类型；
• ReplyType：第二个参数的类型；
• numCalls：后续统计方法调用次数时会用到。

另外，我们还实现了 2 个方法 newArgv 和 newReplyv，用于创建对应类型的实例。

第二步，定义结构体 service：

type service struct {name   stringtyp    reflect.Typercvr   reflect.Valuemethod map[string]*methodType
}

service 的定义非常简洁。name 是映射的结构体的名称，比如 T 、比如 WaitGroup。typ 是结构体的类型。rcvr 是结构体的实例本身，保留 rcvr 是因为调用时需要 rcvr 作为第 0 个参数。method 是 map 类型，存储映射的结构体的所有符合条件的 method。

接下来，完成构造函数newService，入参是任意需要映射为服务的结构体实例：

// newService 创建了一个新的服务实例
func newService(rcvr interface{}) *service {s := new(service)s.rcvr = reflect.ValueOf(rcvr)                  // 获取服务实例的值s.name = reflect.Indirect(s.rcvr).Type().Name() // 获取服务实例的类型名称s.typ = reflect.TypeOf(rcvr)if !ast.IsExported(s.name) {log.Fatalf("rpc server: %s is not a valid service name", s.name)}s.registerMethods() // 调用 registerMethods 方法注册对应服务的方法, registerMethods 在下面实现return s            // 返回一个 service 实例
}// registerMethods 注册了服务的方法
func (s *service) registerMethods() {s.method = make(map[string]*methodType)for i := 0; i < s.typ.NumMethod(); i++ { // 遍历服务的所有方法method := s.typ.Method(i)mType := method.Type// 👇 检查方法的签名是否符合 RPC 方法的规范// 1. 方法必须有三个入参和一个输出参数if mType.NumIn() != 3 || mType.NumOut() != 1 {continue}// 2. 输出参数的类型必须是 error 类型if mType.Out(0) != reflect.TypeOf((*error)(nil)).Elem() {continue}argType, replyType := mType.In(1), mType.In(2)// 3. 参数和返回值的类型必须是导出的或内置类型if !isExportedOrBuiltinType(argType) || !isExportedOrBuiltinType(replyType) {continue}// 经过检查, 符合规定的方法注册到 s.method 当中s.method[method.Name] = &methodType{method:    method,ArgType:   argType,ReplyType: replyType,}// 日志: 输出注册的方法名log.Printf("rpc server: register %s.%s\n", s.name, method.Name)}
}// isExportedOrBuiltinType 的作用是检查类型是否是导出的或内置的类型
func isExportedOrBuiltinType(t reflect.Type) bool {return ast.IsExported(t.Name()) || t.PkgPath() == ""
}

registerMethods 过滤出了符合条件的方法：

• 两个导出或内置类型的入参（反射时为 3 个，第 0 个是自身，类似于 python 的 self，java 中的 this）
• 返回值有且只有一个，类型为 error。

最后，我们还需要实现 call 方法，即，能够通过反射值调用方法：

func (s *service) call(m *methodType, argv, replyv reflect.Value) error {/*s *service: 表示当前服务的实例m *methodType: 表示要调用的方法的信息argv reflect.Value: 表示方法的参数值replyv reflect.Value: 表示方法的返回值(通过指针的方式保存返回值)返回值: error*/atomic.AddUint64(&m.numCalls, 1)		// 原子性地增加调用次数, 保证并发安全f := m.method.Func							// 获取方法的反射函数returnValues := f.Call([]reflect.Value{s.rcvr, argv, replyv})	// 通过反射调用方法if errInter := returnValues[0].Interface(); errInter != nil {	// 处理返回值return errInter.(error)}return nil
}

service 的测试用例

为了保证 service 实现的正确性，我们为 service.go 写了几个测试用例。

定义结构体 Foo，实现 2 个方法，导出方法 Sum 和非导出方法 sum：

package geerpcimport ("fmt""reflect""testing"
)type Foo inttype Args struct{ Num1, Num2 int }func (f Foo) Sum(args Args, reply *int) error {*reply = args.Num1 + args.Num2return nil
}func (f Foo) sum(args Args, reply *int) error {*reply = args.Num1 + args.Num2return nil
}func _assert(condition bool, msg string, v ...interface{}) {if !condition {panic(fmt.Sprintf("assertion failed: "+msg, v...))}
}func TestNewService(t *testing.T) {var foo Foos := newService(&foo)_assert(len(s.method) == 1, "wrong service Method, expect 1, but got %d", len(s.method))mType := s.method["Sum"]_assert(mType != nil, "wrong Method, Sum shouldn't nil")
}func TestMethodType_Call(t *testing.T) {var foo Foos := newService(&foo)mType := s.method["Sum"]argv := mType.newArgv()replyv := mType.newReplyv()argv.Set(reflect.ValueOf(Args{Num1: 1, Num2: 3}))err := s.call(mType, argv, replyv)_assert(err == nil && *replyv.Interface().(*int) == 4 && mType.NumCalls() == 1, "failed to call Foo.Sum")
}

集成到服务端

通过反射结构体已经映射为服务，但请求的处理还没有完成。从接收到请求到回复还差以下几个步骤：

1. 根据入参类型，将请求的 body 反序列化；
2. 调用 service.call ，完成方法调用；
3. 将 reply 序列化为字节流，构造响应报文，返回。

回到代码本身，补全之前在 server.go 中遗留的两个 TODO 任务 readRequest 和 handleRequest 即可。

在此之前，我们还需要在 Server 实现一个 Register 方法，配套实现 findService 方法，即通过 ServiceMethod 从 serviceMap 中找到对应的 service：

// Server represents an RPC Server.
type Server struct {serviceMap sync.Map
}// Register publishes in the server the set of methods
func (server *Server) Register(rcvr interface{}) error {s := newService(rcvr)if _, dup := server.serviceMap.LoadOrStore(s.name, s); dup {return errors.New("rpc: service already defined: " + s.name)}return nil
}// Register publishes the receiver's methods in the DefaultServer
func Register(rcvr interface{}) error { return DefaultServer.Register(rcvr) }func (server *Server) findService(serviceMethod string) (svc *service, mtype *methodType, err error) {dot := strings.LastIndex(serviceMethod, ".")if dot < 0 {err = errors.New("rpc server: service/method request ill-formed" + serviceMethod)return}serviceName, methodName := serviceMethod[:dot], serviceMethod[dot+1:]svci, ok := server.serviceMap.Load(serviceName)if !ok {err = errors.New("rpc server: can't find service " + serviceName)return}svc = svci.(*service)mtype = svc.method[methodName]if mtype == nil {err = errors.New("rpc server: can't find method " + methodName)}return
}

findService看似繁琐，但逻辑非常清晰。ServiceMethod 的构成是“Service.Method”，因此首先将其分割为两部分，第一部分是 Service 的名称，第二部分是方法名。先在 ServiceMap 中找到对应的 service 实例，再从 service 实例的 method 中找到对应的 methodType。

准备工具已经就绪，首先补全 readRequest 方法：

// request stores all information of a call
type request struct {h            *codec.Header // header of requestargv, replyv reflect.Value // argv and replyv of requestmtype        *methodTypesvc          *service
}func (server *Server) readRequest(cc codec.Codec) (*request, error) {h, err := server.readRequestHeader(cc)if err != nil {return nil, err}req := &request{h: h}req.svc, req.mtype, err = server.findService(h.ServiceMethod)if err != nil {return req, err}req.argv = req.mtype.newArgv()req.replyv = req.mtype.newReplyv()// make sure that argvi is a pointer, ReadBody need a pointer as parameterargvi := req.argv.Interface()if req.argv.Type().Kind() != reflect.Ptr {argvi = req.argv.Addr().Interface()}if err = cc.ReadBody(argvi); err != nil {log.Println("rpc server: read argv err:", err)}return req, nil
}

readRequest 方法中最重要的部分，就是通过newArgv() 和 newReplyv() 两个方法创建出两个入参实例，然后通过 cc.ReadBody() 将请求报文反序列化第一个入参 argv，在这里同样需要 argv 可能是值类型，也可能是指针类型，所以处理方式有点差异。

接下来补全 handleRequest 方法：

func (server *Server) handleRequest(cc codec.Codec, req *request, sending *sync.Mutex, wg *sync.WaitGroup) {defer wg.Done()err := req.svc.call(req.mtype, req.argv, req.replyv)if err != nil {req.h.Error = err.Error()server.sendResponse(cc, req.h, invalidRequest, sending)return}server.sendResponse(cc, req.h, req.replyv.Interface(), sending)
}

相对于 readRequest，handleRequest 的实现非常简单，通过req,.svc.call完成方法调用，将 replyv 传递给 sendResponse 完成序列化即可。

到此，我们已经完成了今天的内容，成功在服务端实现了服务注册与调用。

Demo

最后，写一个main 来验证今天的成果：

package mainimport ("Geektutu/GeeRPC/geerpc""log""net""sync""time"
)// Step 1: 定义结构体 Foo 和方法 Sum
type Foo inttype Args struct{ Num1, Num2 int }func (f Foo) Sum(args Args, reply *int) error {*reply = args.Num1 + args.Num2return nil
}// Step 2: 注册 Foo 到 Server 当中, 并启动 RPC 服务
func startServer(addr chan string) {var foo Fooif err := geerpc.Register(&foo); err != nil {log.Fatal("register error:", err)}l, err := net.Listen("tcp", ":8080")if err != nil {log.Fatal("network error:", err)}log.Println("start rpc server on", l.Addr())addr <- l.Addr().String()geerpc.Accept(l)
}func main() {log.SetFlags(0)addr := make(chan string)go startServer(addr)client, _ := geerpc.Dial("tcp", <-addr)defer func() { _ = client.Close() }()time.Sleep(time.Second)// send request & receive responsevar wg sync.WaitGroupfor i := 0; i < 5; i++ {wg.Add(1)go func(i int) {defer wg.Done()args := &Args{Num1: i,Num2: i * i,}var reply intif err := client.Call("Foo.Sum", args, &reply); err != nil {log.Fatal("call Foo.Sum error:", err)}log.Printf("%d + %d = %d", args.Num1, args.Num2, reply)}(i)}wg.Wait()
}// OUTPUT:
rpc server: register Foo.Sum
start rpc server on [::]:8080
3 + 9 = 12
0 + 0 = 0
1 + 1 = 2
4 + 16 = 20
2 + 4 = 6