【golang高级开发】如何使用golang实现自定义RPC框架 RPC (Remote Procedure Call)是一种远程调用协议,通过网络传输,使得程序能够像本地调用一样调用远程服务。在现代微服务架构中,RPC协议被广泛使用。golang通过标准库的net/rpc包提供了一套RPC框架,但是这个框架无法满足一些特定的业务需求,本文就来介绍如何使用golang自己实现一个RPC框架。 1. 基本概念 在实现自定义RPC框架之前,需要先了解以下几个基本概念: - Service:RPC调用的服务,即提供RPC服务的函数集合。 - Method:Service中的方法,即具体的RPC调用方法。 - Codec:序列化和反序列化的方法,将调用的参数和返回值序列化成二进制数据,以便通过网络传输。 - Transport:网络传输协议,用于将序列化后的二进制数据通过网络传输到远程服务。 2. 实现步骤 接下来我们就来实现一个简单的自定义RPC框架,步骤如下: - 定义Service和Method - 实现Codec - 实现Transport - 完成框架 2.1 定义Service和Method 我们以一个简单的计算器服务为例,在服务端提供两个方法Add和Multiply,客户端可以通过RPC调用这两个方法。 定义服务: ```go // 定义CalculatorService接口 type CalculatorService interface { Add(int, int) int Multiply(int, int) int } // 实现具体的CalculatorService type CalculatorServiceImpl struct {} func (c *CalculatorServiceImpl) Add(a, b int) int { return a + b } func (c *CalculatorServiceImpl) Multiply(a, b int) int { return a * b } ``` 定义Service和Method之后,接下来需要定义一个struct来存储Service和其对应的Method。同时,定义一个Register方法,用于注册新的Service和Method。 ```go type Server struct { services map[string]*service } type service struct { typ reflect.Type method map[string]*methodType } type methodType struct { method reflect.Method ArgType reflect.Type ReplyType reflect.Type } func (s *Server) Register(receiver interface{}) error { service := new(service) service.typ = reflect.TypeOf(receiver).Elem() service.method = make(map[string]*methodType) for i := 0; i < service.typ.NumMethod(); i++ { method := service.typ.Method(i) mType := method.Type if mType.NumIn() != 3 || mType.NumOut() != 1 { continue } argType := mType.In(1) replyType := mType.In(2) if !isExportedOrBuiltinType(argType) || !isExportedOrBuiltinType(replyType) { continue } service.method[method.Name] = &methodType{ method: method, ArgType: argType, ReplyType: replyType, } } s.services[reflect.Indirect(reflect.ValueOf(receiver)).Type().Name()] = service return nil } func isExportedOrBuiltinType(t reflect.Type) bool { pkgPath := t.PkgPath() return pkgPath == "" || pkgPath == "builtin" } ``` 在Register方法中,循环遍历service.typ中的所有方法,将满足条件的方法添加到service.method中。最后将service添加到Server.services中。 2.2 实现Codec Codec用于将调用的参数和返回值序列化成二进制数据,以便通过网络传输。 在这里,我们使用golang的标准库encoding/gob实现Codec。Gob是golang标准库中的编解码库,支持任意类型的编解码和传输,比JSON和XML更高效。在实现Codec之前,需要先定义一个request结构体和response结构体,用于存储调用信息和返回信息。 ```go type request struct { ServiceMethod string // 形如"Service.Method" Seq uint64 // 请求序列号 Args []byte // 客户端传递的参数 } type response struct { Seq uint64 // 请求序列号 ServiceMethod string // 形如"Service.Method" Error string // 存储错误信息 Reply []byte // 存储响应参数 } ``` 接下来实现Codec,具体实现代码如下: ```go type Codec struct { conn io.ReadWriteCloser dec *gob.Decoder enc *gob.Encoder mutex sync.Mutex ids uint64 pending map[uint64]*call } type call struct { req *request resp *response done chan *call } func (c *Codec) WriteRequest(method string, args interface{}) (uint64, error) { c.mutex.Lock() defer c.mutex.Unlock() id := c.ids c.ids++ req := &request{ ServiceMethod: method, Seq: id, } buf := bytes.NewBuffer(nil) enc := gob.NewEncoder(buf) if err := enc.Encode(args); err != nil { return 0, err } req.Args = buf.Bytes() call := &call{ req: req, resp: new(response), done: make(chan *call), } c.pending[id] = call if err := c.enc.Encode(req); err != nil { delete(c.pending, id) return 0, err } return id, nil } func (c *Codec) ReadResponseHeader() (*rpc.Response, error) { c.mutex.Lock() defer c.mutex.Unlock() var resp response if err := c.dec.Decode(&resp); err != nil { return nil, err } call := c.pending[resp.Seq] delete(c.pending, resp.Seq) call.resp = &resp call.done <- call return &rpc.Response{ ServiceMethod: resp.ServiceMethod, Seq: resp.Seq, Error: errors.New(resp.Error), }, nil } func (c *Codec) ReadResponseBody(x interface{}) error { c.mutex.Lock() defer c.mutex.Unlock() call := <-c.pending[resp.Seq].done if call.resp.Error != "" { return errors.New(call.resp.Error) } dec := gob.NewDecoder(bytes.NewBuffer(call.resp.Reply)) return dec.Decode(x) } ``` 在上面的代码中,我们使用了一个pending map来存储请求的序列号和请求的返回值。在WriteRequest方法中,我们将请求信息编码成二进制数据,然后将请求信息和该请求的channel存储到pending中。在ReadResponseHeader和ReadResponseBody方法中,我们根据pending中的请求序列号获取该请求对应的call,然后将call.resp进行解码后返回。 2.3 实现Transport Transport用于将序列化后的二进制数据通过网络传输到远程服务。 在golang中,可以使用net包来实现简单的Socket编程。在这里,我们通过net.Dial建立连接后,将Codec中序列化后的数据通过Socket发送到远程服务端。 ```go type Transport struct { conn io.ReadWriteCloser } func (t *Transport) Dial(network, address string) error { conn, err := net.Dial(network, address) if err != nil { return err } t.conn = conn return nil } func (t *Transport) Close() error { return t.conn.Close() } func (t *Transport) Codec() rpc.ClientCodec { return &Codec{ conn: t.conn, dec: gob.NewDecoder(t.conn), enc: gob.NewEncoder(t.conn), pending: make(map[uint64]*call), } } ``` 2.4 完成框架 最后,我们完成自定义RPC框架的实现。具体代码如下: ```go type Server struct { services map[string]*service } type service struct { typ reflect.Type method map[string]*methodType } type methodType struct { method reflect.Method ArgType reflect.Type ReplyType reflect.Type } func (s *Server) Register(receiver interface{}) error { service := new(service) service.typ = reflect.TypeOf(receiver).Elem() service.method = make(map[string]*methodType) for i := 0; i < service.typ.NumMethod(); i++ { method := service.typ.Method(i) mType := method.Type if mType.NumIn() != 3 || mType.NumOut() != 1 { continue } argType := mType.In(1) replyType := mType.In(2) if !isExportedOrBuiltinType(argType) || !isExportedOrBuiltinType(replyType) { continue } service.method[method.Name] = &methodType{ method: method, ArgType: argType, ReplyType: replyType, } } s.services[reflect.Indirect(reflect.ValueOf(receiver)).Type().Name()] = service return nil } func (s *Server) ServeCodec(codec rpc.ServerCodec) error { for { req, err := codec.ReadRequestHeader() if err != nil { if err != io.EOF && err != io.ErrUnexpectedEOF { log.Println("rpc server:", err) } return err } serviceMethod := req.ServiceMethod dot := strings.LastIndex(serviceMethod, ".") if dot < 0 { err := errors.New("rpc server: service/method request ill-formed: " + serviceMethod) log.Println(err.Error()) resp := &rpc.Response{ ServiceMethod: serviceMethod, Seq: req.Seq, Error: err.Error(), } codec.WriteResponse(resp, nil) continue } serviceName, methodName := serviceMethod[:dot], serviceMethod[dot+1:] service, ok := s.services[serviceName] if !ok { err := errors.New("rpc server: can't find service " + serviceName) log.Println(err.Error()) resp := &rpc.Response{ ServiceMethod: serviceMethod, Seq: req.Seq, Error: err.Error(), } codec.WriteResponse(resp, nil) continue } mtype, ok := service.method[methodName] if !ok { err := errors.New("rpc server: can't find method " + methodName) log.Println(err.Error()) resp := &rpc.Response{ ServiceMethod: serviceMethod, Seq: req.Seq, Error: err.Error(), } codec.WriteResponse(resp, nil) continue } argv := reflect.New(mtype.ArgType) replyv := reflect.New(mtype.ReplyType).Elem() if err = codec.ReadRequestBody(argv.Interface()); err != nil { log.Println("rpc server: ", err) resp := &rpc.Response{ ServiceMethod: serviceMethod, Seq: req.Seq, Error: err.Error(), } codec.WriteResponse(resp, nil) continue } // Call the service method. returnValues := mtype.method.Func.Call([]reflect.Value{ reflect.ValueOf(service), reflect.ValueOf(argv.Interface()), replyv, }) // The return value for the method is an error. errInter := returnValues[0].Interface() if errInter != nil { err := errInter.(error) log.Println("rpc server: ", err) resp := &rpc.Response{ ServiceMethod: serviceMethod, Seq: req.Seq, Error: err.Error(), } codec.WriteResponse(resp, nil) continue } resp := &rpc.Response{ ServiceMethod: serviceMethod, Seq: req.Seq, Error: "", } if err = codec.WriteResponse(resp, replyv.Interface()); err != nil { log.Println("rpc server: ", err) } } } func (s *Server) ServeTransport(transport *Transport) error { codec := transport.Codec() defer transport.Close() return s.ServeCodec(codec) } func isExportedOrBuiltinType(t reflect.Type) bool { pkgPath := t.PkgPath() return pkgPath == "" || pkgPath == "builtin" } ``` 在上面的代码中,我们定义了一个Server结构体,用于注册Service和Method,同时实现ServeCodec和ServeTransport方法,用于在服务端处理RPC请求。 3. 测试 完成自定义RPC框架的实现之后,我们需要对其进行测试。下面我们将分别在服务端和客户端使用该RPC框架。 服务端代码: ```go func main() { server := new(Server) server.services = make(map[string]*service) _ = server.Register(&CalculatorServiceImpl{}) transport := new(Transport) _ = transport.Dial("tcp", "localhost:8080") defer transport.Close() log.Fatal(server.ServeTransport(transport)) } ``` 在服务端,我们首先通过Server.Register方法注册了一个CalculatorServiceImpl服务,然后使用Transport.Dial方法连接到特定的地址。 客户端代码: ```go func main() { transport := new(Transport) _ = transport.Dial("tcp", "localhost:8080") defer transport.Close() client := rpc.NewClientWithCodec(transport.Codec()) var res int err := client.Call("CalculatorService.Add", []int{10, 20}, &res) if err != nil { log.Fatal(err) } log.Printf("Add(10, 20) = %d", res) var mul int err = client.Call("CalculatorService.Multiply", []int{10, 20}, &mul) if err != nil { log.Fatal(err) } log.Printf("Multiply(10, 20) = %d", mul) } ``` 在客户端,我们首先通过Transport.Dial方法连接到服务端,然后通过rpc.NewClientWithCodec方法创建一个客户端,并使用client.Call方法调用服务端的方法。 最后,我们启动服务端和客户端,可以看到客户端成功调用了服务端提供的Add和Multiply方法。 4. 总结 本文介绍了如何使用golang实现自定义RPC框架,包括定义Service和Method,实现Codec和Transport,完成框架等步骤,并通过一个简单的计算器服务对该RPC框架进行了测试。该自定义RPC框架适用于一些特定的业务需求,可以满足不同的RPC调用场景。