1.1.1. RPC简介

  • 远程过程调用（Remote Procedure Call，RPC）是一个计算机通信协议
  • 该协议允许运行于一台计算机的程序调用另一台计算机的子程序，而程序员无需额外地为这个交互作用编程

  • 如果涉及的软件采用面向对象编程，那么远程过程调用亦可称作远程调用或远程方法调用

    1.1.2. 流行RPC框架的对比

    RPC - 图1

    1.1.3. golang中如何实现RPC

  • golang中实现RPC非常简单，官方提供了封装好的库，还有一些第三方的库

  • golang官方的net/rpc库使用encoding/gob进行编解码，支持tcp和http数据传输方式，由于其他语言不支持gob编解码方式，所以golang的RPC只支持golang开发的服务器与客户端之间的交互
  • 官方还提供了net/rpc/jsonrpc库实现RPC方法，jsonrpc采用JSON进行数据编解码，因而支持跨语言调用，目前jsonrpc库是基于tcp协议实现的，暂不支持http传输方式
  • 例题：golang实现RPC程序，实现求矩形面积和周长

服务端

  1. package main
  3. import (
  4.     "log"
  5.     "net/http"
  6.     "net/rpc"
  7. )
  9. //    例题：golang实现RPC程序，实现求矩形面积和周长
  11. type Params struct {
  12.     Width, Height int
  13. }
  15. type Rect struct{}
  17. // RPC服务端方法，求矩形面积
  18. func (r *Rect) Area(p Params, ret *int) error {
  19.     *ret = p.Height * p.Width
  20.     return nil
  21. }
  23. // 周长
  24. func (r *Rect) Perimeter(p Params, ret *int) error {
  25.     *ret = (p.Height + p.Width) * 2
  26.     return nil
  27. }
  29. // 主函数
  30. func main() {
  31.     // 1.注册服务
  32.     rect := new(Rect)
  33.     // 注册一个rect的服务
  34.     rpc.Register(rect)
  35.     // 2.服务处理绑定到http协议上
  36.     rpc.HandleHTTP()
  37.     // 3.监听服务
  38.     err := http.ListenAndServe(":8000", nil)
  39.     if err != nil {
  40.         log.Panicln(err)
  41.     }
  42. }

客户端

  1. package main
  3. import (
  4.     "fmt"
  5.     "log"
  6.     "net/rpc"
  7. )
  9. // 传的参数
  10. type Params struct {
  11.     Width, Height int
  12. }
  14. // 主函数
  15. func main() {
  16.     // 1.连接远程rpc服务
  17.     conn, err := rpc.DialHTTP("tcp", ":8000")
  18.     if err != nil {
  19.         log.Fatal(err)
  20.     }
  21.     // 2.调用方法
  22.     // 面积
  23.     ret := 0
  24.     err2 := conn.Call("Rect.Area", Params{50, 100}, &ret)
  25.     if err2 != nil {
  26.         log.Fatal(err2)
  27.     }
  28.     fmt.Println("面积：", ret)
  29.     // 周长
  30.     err3 := conn.Call("Rect.Perimeter", Params{50, 100}, &ret)
  31.     if err3 != nil {
  32.         log.Fatal(err3)
  33.     }
  34.     fmt.Println("周长：", ret)
  35. }
  • golang写RPC程序，必须符合4个基本条件，不然RPC用不了
    • 结构体字段首字母要大写，可以别人调用
    • 函数名必须首字母大写
    • 函数第一参数是接收参数，第二个参数是返回给客户端的参数，必须是指针类型
    • 函数还必须有一个返回值error
  • 练习：模仿前面例题，自己实现RPC程序，服务端接收2个参数，可以做乘法运算，也可以做商和余数的运算，客户端进行传参和访问，得到结果如下：

服务端代码：

  1. package main
  3. import (
  4.    "errors"
  5.    "log"
  6.    "net/http"
  7.    "net/rpc"
  8. )
  10. // 结构体，用于注册的
  11. type Arith struct{}
  13. // 声明参数结构体
  14. type ArithRequest struct {
  15.    A, B int
  16. }
  18. // 返回给客户端的结果
  19. type ArithResponse struct {
  20.    // 乘积
  21.    Pro int
  22.    // 商
  23.    Quo int
  24.    // 余数
  25.    Rem int
  26. }
  28. // 乘法
  29. func (this *Arith) Multiply(req ArithRequest, res *ArithResponse) error {
  30.    res.Pro = req.A * req.B
  31.    return nil
  32. }
  34. // 商和余数
  35. func (this *Arith) Divide(req ArithRequest, res *ArithResponse) error {
  36.    if req.B == 0 {
  37.       return errors.New("除数不能为0")
  38.    }
  39.    // 除
  40.    res.Quo = req.A / req.B
  41.    // 取模
  42.    res.Rem = req.A % req.B
  43.    return nil
  44. }
  46. // 主函数
  47. func main() {
  48.    // 1.注册服务
  49.    rect := new(Arith)
  50.    // 注册一个rect的服务
  51.    rpc.Register(rect)
  52.    // 2.服务处理绑定到http协议上
  53.    rpc.HandleHTTP()
  54.    // 3.监听服务
  55.    err := http.ListenAndServe(":8000", nil)
  56.    if err != nil {
  57.       log.Fatal(err)
  58.    }
  59. }

客户端代码：

  1. package main
  3. import (
  4.    "fmt"
  5.    "log"
  6.    "net/rpc"
  7. )
  9. type ArithRequest struct {
  10.    A, B int
  11. }
  13. // 返回给客户端的结果
  14. type ArithResponse struct {
  15.    // 乘积
  16.    Pro int
  17.    // 商
  18.    Quo int
  19.    // 余数
  20.    Rem int
  21. }
  23. func main() {
  24.    conn, err := rpc.DialHTTP("tcp", ":8000")
  25.    if err != nil {
  26.       log.Fatal(err)
  27.    }
  28.    req := ArithRequest{9, 2}
  29.    var res ArithResponse
  30.    err2 := conn.Call("Arith.Multiply", req, &res)
  31.    if err2 != nil {
  32.       log.Fatal(err2)
  33.    }
  34.    fmt.Printf("%d * %d = %d\n", req.A, req.B, res.Pro)
  35.    err3 := conn.Call("Arith.Divide", req, &res)
  36.    if err3 != nil {
  37.       log.Fatal(err3)
  38.    }
  39.    fmt.Printf("%d / %d 商 %d，余数 = %d\n", req.A, req.B, res.Quo, res.Rem)
  40. }

另外，net/rpc/jsonrpc库通过json格式编解码，支持跨语言调用
服务端代码：

  1. package main
  3. import (
  4.     "fmt"
  5.     "log"
  6.     "net"
  7.     "net/rpc"
  8.     "net/rpc/jsonrpc"
  9. )
  11. type Params struct {
  12.     Width, Height int
  13. }
  14. type Rect struct {
  15. }
  17. func (r *Rect) Area(p Params, ret *int) error {
  18.     *ret = p.Width * p.Height
  19.     return nil
  20. }
  21. func (r *Rect) Perimeter(p Params, ret *int) error {
  22.     *ret = (p.Height + p.Width) * 2
  23.     return nil
  24. }
  25. func main() {
  26.     rpc.Register(new(Rect))
  27.     lis, err := net.Listen("tcp", ":8080")
  28.     if err != nil {
  29.         log.Panicln(err)
  30.     }
  31.     for {
  32.         conn, err := lis.Accept()
  33.         if err != nil {
  34.             continue
  35.         }
  36.         go func(conn net.Conn) {
  37.             fmt.Println("new client")
  38.             jsonrpc.ServeConn(conn)
  39.         }(conn)
  40.     }
  41. }

客户端代码：

  1. package main
  3. import (
  4.     "fmt"
  5.     "log"
  6.     "net/rpc/jsonrpc"
  7. )
  9. type Params struct {
  10.     Width, Height int
  11. }
  13. func main() {
  14.     conn, err := jsonrpc.Dial("tcp", ":8080")
  15.     if err != nil {
  16.         log.Panicln(err)
  17.     }
  18.     ret := 0
  19.     err2 := conn.Call("Rect.Area", Params{50, 100}, &ret)
  20.     if err2 != nil {
  21.         log.Panicln(err2)
  22.     }
  23.     fmt.Println("面积：", ret)
  24.     err3 := conn.Call("Rect.Perimeter", Params{50, 100}, &ret)
  25.     if err3 != nil {
  26.         log.Panicln(err3)
  27.     }
  28.     fmt.Println("周长：", ret)
  29. }

1.1.4. RPC调用流程

  • 微服务架构下数据交互一般是对内 RPC，对外 REST
  • 将业务按功能模块拆分到各个微服务，具有提高项目协作效率、降低模块耦合度、提高系统可用性等优点，但是开发门槛比较高，比如 RPC 框架的使用、后期的服务监控等工作
  • 一般情况下，我们会将功能代码在本地直接调用，微服务架构下，我们需要将这个函数作为单独的服务运行，客户端通过网络调用

1.1.5. 网络传输数据格式

  • 两端要约定好数据包的格式
  • 成熟的RPC框架会有自定义传输协议，这里网络传输格式定义如下，前面是固定长度消息头，后面是变长消息体

RPC - 图2

  • 自己定义数据格式的读写 ```go package rpc

import ( “encoding/binary” “io” “net” )

// 测试网络中读写数据的情况

// 会话连接的结构体 type Session struct { conn net.Conn }

// 构造方法 func NewSession(conn net.Conn) *Session { return &Session{conn: conn} }

// 向连接中去写数据 func (s *Session) Write(data []byte) error { // 定义写数据的格式 // 4字节头部 + 可变体的长度 buf := make([]byte, 4+len(data)) // 写入头部，记录数据长度 binary.BigEndian.PutUint32(buf[:4], uint32(len(data))) // 将整个数据，放到4后边 copy(buf[4:], data) _, err := s.conn.Write(buf) if err != nil { return err } return nil }

// 从连接读数据 func (s *Session) Read() ([]byte, error) { // 读取头部记录的长度 header := make([]byte, 4) // 按长度读取消息 , err := io.ReadFull(s.conn, header) if err != nil { return nil, err } // 读取数据 dataLen := binary.BigEndian.Uint32(header) data := make([]byte, dataLen) , err = io.ReadFull(s.conn, data) if err != nil { return nil, err } return data, nil }

  1. **测试类**
  2. ```go
  3. package rpc
  5. import (
  6.     "fmt"
  7.     "net"
  8.     "sync"
  9.     "testing"
  10. )
  12. func TestSession_ReadWriter(t *testing.T) {
  13.     // 定义地址
  14.     addr := "127.0.0.1:8000"
  15.     my_data := "hello"
  16.     // 等待组定义
  17.     wg := sync.WaitGroup{}
  18.     wg.Add(2)
  19.     // 写数据的协程
  20.     go func() {
  21.         defer wg.Done()
  22.         lis, err := net.Listen("tcp", addr)
  23.         if err != nil {
  24.             t.Fatal(err)
  25.         }
  26.         conn, _ := lis.Accept()
  27.         s := Session{conn: conn}
  28.         err = s.Write([]byte(my_data))
  29.         if err != nil {
  30.             t.Fatal(err)
  31.         }
  32.     }()
  34.     // 读数据的协程
  35.     go func() {
  36.         defer wg.Done()
  37.         conn, err := net.Dial("tcp", addr)
  38.         if err != nil {
  39.             t.Fatal(err)
  40.         }
  41.         s := Session{conn: conn}
  42.         data, err := s.Read()
  43.         if err != nil {
  44.             t.Fatal(err)
  45.         }
  46.         // 最后一层校验
  47.         if string(data) != my_data {
  48.             t.Fatal(err)
  49.         }
  50.         fmt.Println(string(data))
  51.     }()
  52.     wg.Wait()
  53. }

编码解码

  1. package rpc
  3. import (
  4.     "bytes"
  5.     "encoding/gob"
  6. )
  8. // 定义RPC交互的数据结构
  9. type RPCData struct {
  10.     // 访问的函数
  11.     Name string
  12.     // 访问时的参数
  13.     Args []interface{}
  14. }
  16. // 编码
  17. func encode(data RPCData) ([]byte, error) {
  18.     //得到字节数组的编码器
  19.     var buf bytes.Buffer
  20.     bufEnc := gob.NewEncoder(&buf)
  21.     // 编码器对数据编码
  22.     if err := bufEnc.Encode(data); err != nil {
  23.         return nil, err
  24.     }
  25.     return buf.Bytes(), nil
  26. }
  28. // 解码
  29. func decode(b []byte) (RPCData, error) {
  30.     buf := bytes.NewBuffer(b)
  31.     // 得到字节数组解码器
  32.     bufDec := gob.NewDecoder(buf)
  33.     // 解码器对数据节码
  34.     var data RPCData
  35.     if err := bufDec.Decode(&data); err != nil {
  36.         return data, err
  37.     }
  38.     return data, nil
  39. }

1.1.6. 实现RPC服务端

  • 服务端接收到的数据需要包括什么？
    • 调用的函数名、参数列表，还有一个返回值error类型
  • 服务端需要解决的问题是什么？
    • Map维护客户端传来调用函数，服务端知道去调谁
  • 服务端的核心功能有哪些？
    • 维护函数map
    • 客户端传来的东西进行解析
    • 函数的返回值打包，传给客户端 ```go package rpc

import ( “fmt” “net” “reflect” )

// 声明服务端 type Server struct { // 地址 addr string // map 用于维护关系的 funcs map[string]reflect.Value }

// 构造方法 func NewServer(addr string) *Server { return &Server{addr: addr, funcs: make(map[string]reflect.Value)} }

// 服务端需要一个注册Register // 第一个参数函数名，第二个传入真正的函数 func (s *Server) Register(rpcName string, f interface{}) { // 维护一个map // 若map已经有键了 if _, ok := s.funcs[rpcName]; ok { return } // 若map中没值，则将映射加入map，用于调用 fVal := reflect.ValueOf(f) s.funcs[rpcName] = fVal }

// 服务端等待调用的方法 func (s *Server) Run() { // 监听 lis, err := net.Listen(“tcp”, s.addr) if err != nil { fmt.Printf(“监听 %s err :%v”, s.addr, err) return } for { // 服务端循环等待调用 conn, err := lis.Accept() if err != nil { return } serSession := NewSession(conn) // 使用RPC方式读取数据 b, err := serSession.Read() if err != nil { return } // 数据解码 rpcData, err := decode(b) if err != nil { return } // 根据读到的name，得到要调用的函数 f, ok := s.funcs[rpcData.Name] if !ok { fmt.Println(“函数 %s 不存在”, rpcData.Name) return } // 遍历解析客户端传来的参数,放切片里 inArgs := make([]reflect.Value, 0, len(rpcData.Args)) for , arg := range rpcData.Args { inArgs = append(inArgs, reflect.ValueOf(arg)) } // 反射调用方法 // 返回Value类型，用于给客户端传递返回结果,out是所有的返回结果 out := f.Call(inArgs) // 遍历out ，用于返回给客户端，存到一个切片里 outArgs := make([]interface{}, 0, len(out)) for , o := range out { outArgs = append(outArgs, o.Interface()) } // 数据编码，返回给客户端 respRPCData := RPCData{rpcData.Name, outArgs} bytes, err := encode(respRPCData) if err != nil { return } // 将服务端编码后的数据，写出到客户端 err = serSession.Write(bytes) if err != nil { return } } }

  1. <a name="AHhq8"></a>
  2. ### <br />1.1.7. 实现RPC客户端
  4. - 客户端只有函数原型，使用reflect.MakeFunc() 可以完成原型到函数的调用
  5. - reflect.MakeFunc()是Client从函数原型到网络调用的关键
  6. ```go
  7. package rpc
  9. import (
  10.     "net"
  11.     "reflect"
  12. )
  14. // 声明服务端
  15. type Client struct {
  16.     conn net.Conn
  17. }
  19. // 构造方法
  20. func NewClient(conn net.Conn) *Client {
  21.     return &Client{conn: conn}
  22. }
  24. // 实现通用的RPC客户端
  25. // 传入访问的函数名
  26. // fPtr指向的是函数原型
  27. //var select fun xx(User)
  28. //cli.callRPC("selectUser",&select)
  29. func (c *Client) callRPC(rpcName string, fPtr interface{}) {
  30.     // 通过反射，获取fPtr未初始化的函数原型
  31.     fn := reflect.ValueOf(fPtr).Elem()
  32.     // 需要另一个函数，作用是对第一个函数参数操作
  33.     f := func(args []reflect.Value) []reflect.Value {
  34.         // 处理参数
  35.         inArgs := make([]interface{}, 0, len(args))
  36.         for _, arg := range args {
  37.             inArgs = append(inArgs, arg.Interface())
  38.         }
  39.         // 连接
  40.         cliSession := NewSession(c.conn)
  41.         // 编码数据
  42.         reqRPC := RPCData{Name: rpcName, Args: inArgs}
  43.         b, err := encode(reqRPC)
  44.         if err != nil {
  45.             panic(err)
  46.         }
  47.         // 写数据
  48.         err = cliSession.Write(b)
  49.         if err != nil {
  50.             panic(err)
  51.         }
  52.         // 服务端发过来返回值，此时应该读取和解析
  53.         respBytes, err := cliSession.Read()
  54.         if err != nil {
  55.             panic(err)
  56.         }
  57.         // 解码
  58.         respRPC, err := decode(respBytes)
  59.         if err != nil {
  60.             panic(err)
  61.         }
  62.         // 处理服务端返回的数据
  63.         outArgs := make([]reflect.Value, 0, len(respRPC.Args))
  64.         for i, arg := range respRPC.Args {
  65.             // 必须进行nil转换
  66.             if arg == nil {
  67.                 // reflect.Zero()会返回类型的零值的value
  68.                 // .out()会返回函数输出的参数类型
  69.                 outArgs = append(outArgs, reflect.Zero(fn.Type().Out(i)))
  70.                 continue
  71.             }
  72.             outArgs = append(outArgs, reflect.ValueOf(arg))
  73.         }
  74.         return outArgs
  75.     }
  76.     // 完成原型到函数调用的内部转换
  77.     // 参数1是reflect.Type
  78.     // 参数2 f是函数类型，是对于参数1 fn函数的操作
  79.     // fn是定义，f是具体操作
  80.     v := reflect.MakeFunc(fn.Type(), f)
  81.     // 为函数fPtr赋值，过程
  82.     fn.Set(v)
  83. }


1.1.8. 实现RPC通信测试

  • 给服务端注册一个查询用户的方法，客户端使用RPC方式调用 ```go package rpc

import ( “encoding/gob” “fmt” “net” “testing” )

//  给服务端注册一个查询用户的方法，客户端使用RPC方式调用

// 定义用户对象 type User struct { Name string Age int }

// 用于测试用户查询的方法 func queryUser(uid int) (User, error) { user := make(map[int]User) // 假数据 user[0] = User{“zs”, 20} user[1] = User{“ls”, 21} user[2] = User{“ww”, 22} // 模拟查询用户 if u, ok := user[uid]; ok { return u, nil } return User{}, fmt.Errorf(“%d err”, uid) }

func TestRPC(t *testing.T) { // 编码中有一个字段是interface{}时，要注册一下 gob.Register(User{}) addr := “127.0.0.1:8000” // 创建服务端 srv := NewServer(addr) // 将服务端方法，注册一下 srv.Register(“queryUser”, queryUser) // 服务端等待调用 go srv.Run() // 客户端获取连接 conn, err := net.Dial(“tcp”, addr) if err != nil { fmt.Println(“err”) } // 创建客户端对象 cli := NewClient(conn) // 需要声明函数原型 var query func(int) (User, error) cli.callRPC(“queryUser”, &query) // 得到查询结果 u, err := query(1) if err != nil { fmt.Println(“err”) } fmt.Println(u) } ```