Golang——RPC

一. RPC简介

• 长途过程调用(Remote Procedure Call，RPC)是一个盘算机通信协议。
  
• 该协议运行于一台盘算机的程序调用别的一台盘算机的子程序，而程序员无需额外的为这个交互作用编程。
  
• 如果涉及的软件采用面向对象编程，那么长途过程调用亦可称作长途调用或长途方法调用。
  
• 长途过程调用是一个分布式盘算的客户端-服务器(Client/Server)的例子。
  
• 长途过程调用总是由客户端对服务器发出一个实行多少过程请求，并由客户端提供参数，实行效果将返回客户端。由于存在各式各样的变体和差异，对应的派生了各式长途过程调用协议，而他们并不相互兼容。
  
• RPC本身是client-server模型，也是一种request-response协议。有些实现扩展了长途调用的模型。实现了双向的服务调用，但是不管怎么样，调用过程还是由一个客户端发起，服务端提供响应。
  
• 调用过程为：
  
  
  
  • client调用client stub，这是一次本地过程调用
    
  • client stub将参数打包成一个消息，然后发送这个消息
    
  • client地点的系统将消息发送给server
    
  • server的系统将收到的包传给server stub
    
  • server stub解包得到参数
    
  • 末了server stub调用服务过程，返回效果按照相反的步骤传给client
    
  
  

二. 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.         "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. }

复制代码

• golang写RPC程序必须符合4个基本条件，不然RPC用不了
  
  
  
  • 结构体字段首字母必须大写，可以别人调用
    
  • 函数名必须首字母大写
    
  • 函数(服务方法)第一参数是担当参数，第二个参数是返回客户端的参数，必须是指针范例
    
  • 函数(服务方法)还必须有一个返回值error
    
  
  

   别的，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，对外REST
  
• 将业务按功能模块拆分到各个微服务，具有进步项目协作效率，降低模块耦合度，进步系统可用性等优点
  
• 一般情况下，我们将功能代码在本地直接调用，微服务框架下，我们需要将这个函数作为单独的服务运行，客户端通过网络调用。
  

我理解的微服务(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服务端

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

• 服务端接收的数据包罗
  
  
  
  • 调用的函数名，参数，返回值
    
  
  
• 服务器端需要办理的题目
  
  
  
  • 需要维护一个map，来生存调用的函数
    
  
  
• 服务器的焦点功能
  
  
  
  • 维护函数map
    
  • 接收到客户端的数据，并解析
    
  • 找到调用的函数，将返回值打包穿给客户端
    
  
  

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客户端

• 客户端只有函数原型，利用reflect.MakeFunc()可以完成原型到函数的调用。
  
  
  
  • 即reflect.MakeFunc()函数可以将一个函数范例和函数实现结合起来，通过函数原型调用对应函数实现。
    
  
  

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企服之家，中国第一个企服评测及商务社交产业平台。