ToB企服应用市场:ToB评测及商务社交产业平台

标题: Golang——RPC [打印本页]
作者: tsx81429    时间: 2024-6-15 00:50
标题: Golang——RPC
一. RPC简介


二. golang中如何实现RPC


例题：golang实现RPC程序，实现求矩形面积和周长。
服务器：
        注册服务，监听请求，处理请求。
  1. package main
  3. import (
  4.         "fmt"
  5.         "log"
  6.         "net/http"
  7.         "net/rpc"
  8. )
  10. // 接收参数
  11. type Params struct {
  12.         Length int
  13.         Heigth int
  14. }
  16. // 用于注册
  17. type Rect struct{}
  19. func (r *Rect) Area(p Params, ret *int) error {
  20.         fmt.Println("参数：", p.Heigth, p.Length)
  21.         *ret = p.Heigth * p.Length
  22.         return nil
  23. }
  25. func (r *Rect) Perimeter(p Params, ret *int) error {
  26.         fmt.Println("参数：", p.Heigth, p.Length)
  27.         *ret = 2 * (p.Heigth + p.Length)
  28.         return nil
  29. }
  31. func main() {
  32.         //1. 注册服务
  33.         rect := new(Rect)
  34.         rpc.Register(rect)
  36.         //2. 服务器处理绑定到http协议上
  37.         rpc.HandleHTTP()
  38.         //3. 监听服务
  39.         err := http.ListenAndServe(":8080", nil)
  40.         if err != nil {
  41.                 log.Panicln(err)
  42.         }
  43. }
复制代码
客户端：
        连接rpc服务，发送请求调用方法。
  1. package main
  3. import (
  4.         "fmt"
  5.         "log"
  6.         "net/rpc"
  7. )
  9. // 参数字段要和服务器的一样
  10. type Params struct {
  11.         Length int
  12.         Heigth int
  13. }
  15. func main() {
  16.         conn, err := rpc.DialHTTP("tcp", ":8080")
  17.         if err != nil {
  18.                 log.Fatal(err)
  19.         }
  21.         ret := 0
  22.         p := Params{
  23.                 20,
  24.                 30,
  25.         }
  26.         //调用服务器的方法
  27.         err = conn.Call("Rect.Area", p, &ret)
  28.         if err != nil {
  29.                 log.Fatal(err)
  30.         }
  31.         fmt.Println("面积", ret)
  33.         err = conn.Call("Rect.Perimeter", p, &ret)
  34.         if err != nil {
  35.                 log.Fatal(err)
  36.         }
  37.         fmt.Println("周长", ret)
  38. }
复制代码

   别的，net/rpc/jsonrpc库通过json格式编解码，支持跨语言调用。
  服务端：
  1. package main
  3. import (
  4.         "fmt"
  5.         "log"
  6.         "net"
  7.         "net/rpc"
  8.         "net/rpc/jsonrpc"
  9. )
  11. // 接收参数
  12. type Params struct {
  13.         A int
  14.         B int
  15. }
  17. type Rect struct{}
  19. func (r *Rect) Area(p Params, ret *int) error {
  20.         *ret = p.A * p.B
  21.         return nil
  22. }
  24. func (r *Rect) Perimeter(p Params, ret *int) error {
  25.         *ret = 2 * (p.A + p.B)
  26.         return nil
  27. }
  29. func main() {
  30.         //注册服务
  31.         rpc.Register(new(Rect))
  32.         //监听连接
  33.         l, err := net.Listen("tcp", ":8080")
  34.         if err != nil {
  35.                 log.Fatal(err)
  36.         }
  38.         for {
  39.                 //接收客户端连接
  40.                 conn, err := l.Accept()
  41.                 if err != nil {
  42.                         continue
  43.                 }
  44.                 //创建协程处理请求
  45.                 go func(conn net.Conn) {
  46.                         fmt.Println("get a new client")
  47.                         //只是处理服务，没有监听和接收请求
  48.                         jsonrpc.ServeConn(conn)
  49.                 }(conn)
  50.         }
  51. }
复制代码
客户端：
  1. package main
  3. import (
  4.         "fmt"
  5.         "log"
  6.         "net/rpc/jsonrpc"
  7. )
  9. // 参数字段要和服务器的一样
  10. type Params struct {
  11.         A int
  12.         B int
  13. }
  15. func main() {
  16.         //远程连接rpc服务
  17.         conn, err := jsonrpc.Dial("tcp", ":8080")
  18.         if err != nil {
  19.                 log.Fatal(err)
  20.         }
  21.         //调用方法
  22.         ret := 0
  23.         err = conn.Call("Rect.Area", Params{10, 20}, &ret)
  24.         if err != nil {
  25.                 log.Fatal(err)
  26.         }
  27.         fmt.Println("ret=", ret)
  29.         err = conn.Call("Rect.Perimeter", Params{10, 20}, &ret)
  30.         if err != nil {
  31.                 log.Fatal(err)
  32.         }
  33.         fmt.Println("ret=", ret)
  34. }
复制代码
三.  RPC调用流程


我理解的微服务(RPC)，将项目中须要的功能注册成一个服务，客户端可以直接调用。
四. 网络传输数据格式




  1. package rpc
  3. import (
  4.         "encoding/binary"
  5.         "fmt"
  6.         "io"
  7.         "net"
  8. )
  10. type Session struct {
  11.         Conn net.Conn
  12. }
  14. func (s *Session) Write(data []byte) error {
  15.         //定义写数据格式
  16.         //4字节包头 + 变长数据
  17.         buf := make([]byte, 4+len(data))
  18.         //写入头部
  19.         binary.BigEndian.PutUint32(buf[:4], uint32(len(data)))
  20.         //将有效数据写到头部后面
  21.         copy(buf[4:], data)
  22.         //发送
  23.         _, err := s.Conn.Write(buf)
  24.         if err != nil {
  25.                 return err
  26.         }
  27.         return nil
  28. }
  30. func (s *Session) Read() ([]byte, error) {
  31.         //读取头部
  32.         header := make([]byte, 4)
  33.         //按长度读取数据
  34.         _, err := io.ReadFull(s.Conn, header)
  35.         if err != nil {
  36.                 return nil, err
  37.         }
  39.         //读取数据，报头保存的是有效数据长度
  40.         dataLen := binary.BigEndian.Uint32(header)
  41.         fmt.Println(dataLen)
  42.         data := make([]byte, dataLen)
  44.         _, err = io.ReadFull(s.Conn, data)
  45.         if err != nil {
  46.                 return nil, err
  47.         }
  48.         return data, nil
  49. }
复制代码
测试代码：
  1. package rpc
  3. import (
  4.         "errors"
  5.         "fmt"
  6.         "net"
  7.         "sync"
  8.         "testing"
  9. )
  11. func TestSession_ReadWrite(t *testing.T) {
  12.         //地址和数据
  13.         addr := "127.0.0.1:8080"
  14.         my_data := "hello"
  16.         //等待组
  17.         wg := sync.WaitGroup{}
  18.         wg.Add(2)
  19.         go func() {
  20.                 //监听
  21.                 defer wg.Done()
  22.                 l, err := net.Listen("tcp", addr)
  23.                 if err != nil {
  24.                         t.Fatal(err)
  25.                 }
  26.                 //接收连接
  27.                 conn, err := l.Accept()
  28.                 if err != nil {
  29.                         t.Fatal(err)
  30.                 }
  31.                 //写数据
  32.                 s := Session{Conn: conn}
  33.                 err = s.Write([]byte(my_data))
  34.                 if err != nil {
  35.                         t.Fatal(err)
  36.                 }
  37.         }()
  39.         go func() {
  40.                 defer wg.Done()
  41.                 conn, err := net.Dial("tcp", addr)
  42.                 if err != nil {
  43.                         t.Fatal(err)
  44.                 }
  46.                 s := Session{Conn: conn}
  47.                 data, err := s.Read()
  48.                 if err != nil {
  49.                         t.Fatal(err)
  50.                 }
  52.                 //校验
  53.                 if string(data) != my_data {
  54.                         t.Fatal(errors.New("数据错误"))
  55.                 }
  56.                 fmt.Println(string(data))
  57.         }()
  58.         wg.Wait()
  59. }
复制代码

 编码息争码：
  1. func Encode(data RPCData) ([]byte, error) {
  2.         //得到字节数组编码器
  3.         var buf bytes.Buffer
  4.         buffEnc := gob.NewEncoder(&buf)
  5.         //编码器对数据编码
  6.         if err := buffEnc.Encode(data); err != nil {
  7.                 return nil, err
  8.         }
  10.         return buf.Bytes(), nil
  11. }
  13. func Decode(data []byte) (RPCData, error) {
  14.         buff := bytes.NewBuffer(data)
  15.         //得到字节数组解码器
  16.         buffDec := gob.NewDecoder(buff)
  17.         //解码器对数据解码
  18.         var res RPCData
  19.         if err := buffDec.Decode(&res); err != nil {
  20.                 return res, err
  21.         }
  23.         return res, nil
  24. }
复制代码
五. 实现RPC服务端

        要实现的一个功能为，接收到客户端发过来需要调用的函数和参数，实现对应函数的调用，将效果返回。

  1. package main
  3. import (
  4.         "bytes"
  5.         "encoding/binary"
  6.         "encoding/gob"
  7.         "errors"
  8.         "fmt"
  9.         "io"
  10.         "log"
  11.         "net"
  12.         "reflect"
  13. )
  15. // 定义交互的数据结构
  16. type RPCData struct {
  17.         //访问的函数
  18.         Name string
  19.         //参数
  20.         Args []interface{}
  21. }
  23. func encode(data RPCData) ([]byte, error) {
  24.         //得到字节数组编码器
  25.         var buf bytes.Buffer
  26.         gob.Register(User{})
  27.         buffEnc := gob.NewEncoder(&buf)
  28.         //编码器对数据编码
  29.         if err := buffEnc.Encode(data); err != nil {
  30.                 return nil, err
  31.         }
  33.         return buf.Bytes(), nil
  34. }
  36. func decode(data []byte) (RPCData, error) {
  37.         buff := bytes.NewBuffer(data)
  38.         //得到字节数组解码器
  39.         gob.Register(User{})
  40.         buffDec := gob.NewDecoder(buff)
  41.         //解码器对数据解码
  42.         var res RPCData
  43.         if err := buffDec.Decode(&res); err != nil {
  44.                 return res, err
  45.         }
  47.         return res, nil
  48. }
  50. // 发送和接收数据结构
  51. type Session struct {
  52.         Conn net.Conn
  53. }
  55. func NewSession(conn net.Conn) *Session {
  56.         return &Session{Conn: conn}
  57. }
  59. func (s *Session) Write(data []byte) error {
  60.         //定义写数据格式
  61.         //4字节包头 + 变长数据
  62.         buf := make([]byte, 4+len(data))
  63.         //写入头部
  64.         binary.BigEndian.PutUint32(buf[:4], uint32(len(data)))
  65.         //将有效数据写到头部后面
  66.         copy(buf[4:], data)
  67.         //发送
  68.         _, err := s.Conn.Write(buf)
  69.         if err != nil {
  70.                 return err
  71.         }
  72.         return nil
  73. }
  75. func (s *Session) Read() ([]byte, error) {
  76.         //读取头部
  77.         header := make([]byte, 4)
  78.         //按长度读取数据
  79.         _, err := io.ReadFull(s.Conn, header)
  80.         if err != nil {
  81.                 return nil, err
  82.         }
  84.         //读取数据，报头保存的是有效数据长度
  85.         dataLen := binary.BigEndian.Uint32(header)
  86.         fmt.Println(dataLen)
  87.         data := make([]byte, dataLen)
  89.         _, err = io.ReadFull(s.Conn, data)
  90.         if err != nil {
  91.                 return nil, err
  92.         }
  93.         return data, nil
  94. }
  96. // 服务器
  97. type Server struct {
  98.         //访问地址
  99.         addr string
  100.         //维护函数map
  101.         funcs map[string]reflect.Value
  102. }
  104. func NewServer(addr string) *Server {
  105.         return &Server{addr: addr, funcs: make(map[string]reflect.Value)}
  106. }
  108. func (s *Server) Register(key string, value interface{}) error {
  109.         if _, ok := s.funcs[key]; ok {
  110.                 return errors.New(key + "已存在")
  111.         }
  113.         s.funcs[key] = reflect.ValueOf(value)
  114.         return nil
  115. }
  117. func (s *Server) Run() {
  118.         //监听
  119.         ls, err := net.Listen("tcp", s.addr)
  120.         if err != nil {
  121.                 log.Printf("监听addr:%s, 失败", s.addr)
  122.                 return
  123.         }
  125.         for {
  126.                 conn, err := ls.Accept()
  127.                 if err != nil {
  128.                         return
  129.                 }
  131.                 //读数据
  132.                 session := NewSession(conn)
  133.                 data, err := session.Read()
  134.                 if err != nil {
  135.                         return
  136.                 }
  137.                 //解码
  138.                 rpc_data, err := decode(data)
  139.                 if err != nil {
  140.                         return
  141.                 }
  143.                 //查找访问的函数
  144.                 f, ok := s.funcs[rpc_data.Name]
  145.                 if !ok {
  146.                         log.Printf("%s不存在\n", rpc_data.Name)
  147.                         continue
  148.                 }
  149.                 //参数
  150.                 args := make([]reflect.Value, 0, len(rpc_data.Args))
  151.                 for _, v := range rpc_data.Args {
  152.                         if v != nil {
  153.                                 args = append(args, reflect.ValueOf(v))
  154.                         }
  155.                 }
  156.                 //反射调用方法
  157.                 //返回value类型，用于客户端传递返回结果，out是所有返回结果
  158.                 out := f.Call(args)
  159.                 outInters := make([]interface{}, 0, len(out))
  160.                 for _, v := range out {
  161.                         outInters = append(outInters, v.Interface())
  162.                 }
  163.                 fmt.Println("args: ", args, "-", len(args), "-", outInters)
  165.                 //发送给客户端
  166.                 //编码
  167.                 respRpcData := RPCData{Name: rpc_data.Name, Args: outInters}
  168.                 respdata, err := encode(respRpcData)
  169.                 if err != nil {
  170.                         log.Printf("encode fail %v\n", err)
  171.                         continue
  172.                 }
  174.                 err = session.Write(respdata)
  175.                 if err != nil {
  176.                         log.Printf("Write fail %v\n", err)
  177.                         continue
  178.                 }
  179.         }
  181. }
  183. type User struct {
  184.         Name string
  185.         Age  int
  186. }
  188. func queryUser(uid int) (User, error) {
  189.         users := make(map[int]User)
  190.         users[0] = User{"zs", 10}
  191.         users[1] = User{"ls", 20}
  192.         users[2] = User{"ww", 25}
  193.         users[3] = User{"lc", 18}
  194.         fmt.Println(uid)
  195.         if v, ok := users[uid]; ok {
  196.                 fmt.Println(v)
  197.                 return v, nil
  198.         }
  199.         return User{}, errors.New("Not Found")
  200. }
  202. func main() {
  203.         s := NewServer("127.0.0.1:8080")
  205.         //给服务器注册函数
  206.         s.Register("queryUser", queryUser)
  208.         s.Run()
  209. }
复制代码
六. 实现RPC客户端



  1. package main
  3. import (
  4.         "bytes"
  5.         "encoding/binary"
  6.         "encoding/gob"
  7.         "fmt"
  8.         "io"
  9.         "log"
  10.         "net"
  11.         "reflect"
  12. )
  14. type RPCData struct {
  15.         //访问的函数
  16.         Name string
  17.         //参数
  18.         Args []interface{}
  19. }
  21. func encode(data RPCData) ([]byte, error) {
  22.         //得到字节数组编码器
  23.         var buf bytes.Buffer
  24.         //注册接口类型
  25.         gob.Register(User{})
  26.         buffEnc := gob.NewEncoder(&buf)
  27.         //编码器对数据编码
  28.         if err := buffEnc.Encode(data); err != nil {
  29.                 return nil, err
  30.         }
  32.         return buf.Bytes(), nil
  33. }
  35. func decode(data []byte) (RPCData, error) {
  36.         buff := bytes.NewBuffer(data)
  37.         //得到字节数组解码器
  38.         gob.Register(User{})
  39.         buffDec := gob.NewDecoder(buff)
  40.         //解码器对数据解码
  41.         var res RPCData
  42.         if err := buffDec.Decode(&res); err != nil {
  43.                 return res, err
  44.         }
  46.         return res, nil
  47. }
  49. type Session struct {
  50.         Conn net.Conn
  51. }
  53. func NewSession(conn net.Conn) *Session {
  54.         return &Session{Conn: conn}
  55. }
  57. func (s *Session) Write(data []byte) error {
  58.         //定义写数据格式
  59.         //4字节包头 + 变长数据
  60.         buf := make([]byte, 4+len(data))
  61.         //写入头部
  62.         binary.BigEndian.PutUint32(buf[:4], uint32(len(data)))
  63.         //将有效数据写到头部后面
  64.         copy(buf[4:], data)
  65.         //发送
  66.         _, err := s.Conn.Write(buf)
  67.         if err != nil {
  68.                 return err
  69.         }
  70.         return nil
  71. }
  73. func (s *Session) Read() ([]byte, error) {
  74.         //读取头部
  75.         header := make([]byte, 4)
  76.         //按长度读取数据
  77.         _, err := io.ReadFull(s.Conn, header)
  78.         if err != nil {
  79.                 return nil, err
  80.         }
  82.         //读取数据，报头保存的是有效数据长度
  83.         dataLen := binary.BigEndian.Uint32(header)
  84.         fmt.Println(dataLen)
  85.         data := make([]byte, dataLen)
  87.         _, err = io.ReadFull(s.Conn, data)
  88.         if err != nil {
  89.                 return nil, err
  90.         }
  91.         return data, nil
  92. }
  94. // 客户端
  95. type Client struct {
  96.         Conn net.Conn
  97. }
  99. func NewClient(conn net.Conn) *Client {
  100.         return &Client{Conn: conn}
  101. }
  103. // fname为访问服务器函数名
  104. // fptr为函数原型
  105. func (c *Client) callRpc(fname string, fptr interface{}) {
  107.         //获取函数原型
  108.         fnptr := reflect.ValueOf(fptr).Elem()
  109.         //函数实现，发送数据给服务器，收到服务端数据
  110.         //args调用时传进来的参数
  111.         //返回值为得到的结果
  112.         f := func(args []reflect.Value) []reflect.Value {
  113.                 //获得参数
  114.                 argSlice := make([]interface{}, len(args))
  115.                 for _, v := range args {
  116.                         argSlice = append(argSlice, v.Interface())
  117.                 }
  119.                 session := NewSession(c.Conn)
  121.                 //发送到服务器
  122.                 rpcData := RPCData{Name: fname, Args: argSlice}
  123.                 //编码
  124.                 data, err := encode(rpcData)
  125.                 if err != nil {
  126.                         log.Println("encode fail", err)
  127.                         return nil
  128.                 }
  129.                 err = session.Write(data)
  130.                 if err != nil {
  131.                         log.Println("Write fail", err)
  132.                         return nil
  133.                 }
  135.                 //接收服务器响应
  136.                 respData, err := session.Read()
  137.                 if err != nil {
  138.                         log.Println("Read fail", err)
  139.                         return nil
  140.                 }
  141.                 //结果在Args里
  142.                 respRpcData, err := decode(respData)
  143.                 if err != nil {
  144.                         log.Println("decode fail", err)
  145.                         return nil
  146.                 }
  148.                 res := make([]reflect.Value, 0, len(respRpcData.Args))
  149.                 for i, v := range respRpcData.Args {
  150.                         if v == nil {
  151.                                 //因为返回err可能为nil
  152.                                 //进行nil转换
  153.                                 //reflect.Zero会返回类型的零值的value
  154.                                 //.out会返回函数输出的参数类型
  155.                                 res = append(res, reflect.Zero(fnptr.Type().Out(i)))
  156.                         } else {
  157.                                 res = append(res, reflect.ValueOf(v))
  158.                         }
  159.                 }
  160.                 return res
  161.         }
  163.         real_fn := reflect.MakeFunc(fnptr.Type(), f)
  164.         //为函数原型赋值函数实现
  165.         fnptr.Set(real_fn)
  166. }
  168. type User struct {
  169.         Name string
  170.         Age  int
  171. }
  173. func main() {
  174.         conn, err := net.Dial("tcp", ":8080")
  175.         if err != nil {
  176.                 log.Fatal(err)
  177.                 return
  178.         }
  180.         c := NewClient(conn)
  181.         var query func(int) (User, error)
  182.         c.callRpc("queryUser", &query)
  183.     //进行查询
  184.         u, err := query(2)
  185.         if err != nil {
  186.                 log.Fatal(err)
  187.                 return
  188.         }
  189.         fmt.Println(u)
  190. }
复制代码
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。



欢迎光临 ToB企服应用市场:ToB评测及商务社交产业平台 (https://dis.qidao123.com/) Powered by Discuz! X3.4