golang rpc

东湖之滨  金牌会员 | 2024-10-3 02:48:57 | 显示全部楼层 | 阅读模式
打印 上一主题 下一主题
楼主

主题 875|帖子 875|积分 2629

RPC（Remote Procedure Call）长途过程调用，简单的理解是一个节点请求另一个节点提供的服务，对应rpc的是本地过程调用，函数调用是最常用的本地过程调用，将本地过程调用变成长途调用碰面对着各种问题。
以两数相加为例
  1. package main
  3. import "fmt"
  5. func add(a, b int) int {
  6.         return a + b
  7. }
  8. func main() {
  9.         fmt.Println(add(1, 2))
  10. }
复制代码
函数调用过程：
（1）将1和2压入函数的栈中
（2）进入add函数，从栈中取出1和2分别赋值给a和b
（3）实行a+b将效果压栈
（4）将栈中的效果取出打印
长途过程面对的问题
1.本来的本地函数放到另外一个服务器上去运行，但是引入了很多新问题
2.Call的id映射
我们怎么告诉长途呆板我们要调用add，而不是sub或者Foo呢？在本地调用中，函数体是直接通过函数指针来指定的，我们调用add，编译器就自动帮我们调用它相应的函数指针。但是在长途调用中，函数指针是不可的，因为两个进程的地点空间是完全不一样的。以是，在RPC中，所有的函数都必须有自己的一个ID。这个ID在所有进程中都是唯一确定的。客户端在做长途过程调用时，必须附上这个ID。然后我们还需要在客户端和服务端分别维护一个（函数＜–>Call ID｝的对应表。两者的表不一定需要完全类似，但类似的函数对应的Call ID必须类似。当客户端需要进行长途调用时，它就查一下这个表，找出相应的Call ID，然后把它传给服务端，服务端也通过查表，来确定客户端需要调用的函数，然后实行相应函数的代码。
3.序列化和反序列化
客户端怎么把参数值传给长途的函数呢？在本地调用中，我们只需要把参数压到栈里，然后让函数自己去栈里读就行。但是在长途过程调用时，客户端跟服务端是不同的进程，不能通过内存来传递参数。甚至偶然间客户端和服务端使用的都不是同一种语言（比如服务端用C++，客户端用Java或者Python）。这时间就需要客户端把参数先转成一个字节流，传给服务端后，再把字节流转成自己能读取的格式。这个过程叫序列化和反序列化。同理，从服务端返回的值也需要序列化反序列化的过程。
4.网络传输
长途调用每每用在网络上，客户端和服务端是通过网络连接的。所有的数据都需要通过网络传输，因此就需要有一个网络传输层。网络传输层需要把Call ID和序列化后的参数字节流传给服务端，然后再把序列化后的调用效果传给客户端。只要能完成这两者的，都可以作为传输层使用。因此，他所使用的协议其实是不限的，能完成传输就行，尽管大部分RPC框架都使用TCP协议，但其实UDP也可以，而RPC干脆就用了HTTP2。Java的Netty也属于这层的东西。
RPC第一个要点：数据编码协议

一般采用将数据传输到gin，gin传输到服务端，服务端负责剖析数据
客户端流程
  1.         1.建立连接tcp/http
  2.         2.将employee对象序列化成json字符串-序列化
  3.         3.发送json字符串-调用成功后实际上你接受到的是一个二进制的数据
  4.         4.等待服务器发送结果
  5.         5将服务返回的数据解析成PrintResult对象-反序列化
复制代码
服务端流程
  1.         1.监听网络接口80
  2.         2.读取数据-二进制的json数据
  3.         3.对数据进行反序列化Employee对象
  4.         4.开始处理业务逻辑
  5.         5.将处理的结果PrintReuslt发序列化成json二进制数据-序列化
  6.         6.将数据返回
复制代码
序列化和反序列化是可以选择的，不一定要采用json、xml、protobuf、msgpack
RPC第二个要点：传输协议

http协议：http1.x http2.0协议
http协议底层使用的也是tcp，http如今主流的是http1.x，这种协议有性能问题（一次性），一旦效果返回，连接就断开。我们可以直接基于tcp/udp协议去封装一层协议myhttp，没有通用型，http2.0既有http的特性也有长连接的特性（grpc就是基于http2.0的）
http协议是文本协议，http底层的传输协议是tcp。grpc基于http2.0，传输协议也是tcp

http协议具有一次性的问题：一旦对方返回了效果，连接断开，http2.0通过长连接办理了这个问题。
基于Http Server实现rpc请求

server端
  1. package main
  3. import (
  4.         "encoding/json"
  5.         "fmt"
  6.         "net/http"
  7.         "strconv"
  8. )
  10. func add(a, b int) int {
  11.         return a + b
  12. }
  14. func main() {
  15.         //get方法http://127.0.0.1:8000/add?a=1&b=2或http://127.0.0.1:8000?method=add&a=1&b=2
  16.         //返回的格式化：json{"data":3}
  17.         //1、callId的问题：r.URL.Path，2、数据的传输协议：url的参数传输协议，3、网络传输协议：http
  18.         http.HandleFunc("/add", func(w http.ResponseWriter, r *http.Request) {
  19.                 err := r.ParseForm() //解析参数
  20.                 if err != nil {
  21.                         panic("error")
  22.                 }
  23.                 fmt.Println("path:", r.URL.Path)
  24.                 a, err := strconv.Atoi(r.Form["a"][0])
  25.                 if err != nil {
  26.                         panic("transform error")
  27.                 }
  28.                 b, err := strconv.Atoi(r.Form["b"][0])
  29.                 if err != nil {
  30.                         panic("transform error")
  31.                 }
  32.                 w.Header().Set("Content-Type", "application/json")
  33.                 jData, err := json.Marshal(map[string]int{
  34.                         "data": a + b,
  35.                 })
  36.                 w.Write(jData)
  37.         })
  38.         _ = http.ListenAndServe(":8000", nil)
  39. }
复制代码

缺点：http1.x，麻烦，性能不高
客户端：
  1. package main
  3. import (
  4.         "encoding/json"
  5.         "fmt"
  6.         "time"
  8.         "github.com/kirinlabs/HttpRequest"
  9. )
  11. type ResponseData struct {
  12.         Data int `json:"data"`
  13. }
  15. func Add(a, b int) int {
  16.         req := HttpRequest.NewRequest()
  17.         res, _ := req.Get(fmt.Sprintf("http://127.0.0.1:8000/%s?a=%d&b=%d", "add", a, b))
  18.         body, _ := res.Body()
  19.         rspData := ResponseData{}
  20.         _ = json.Unmarshal(body, &rspData)
  21.         return rspData.Data
  22. }
  23. func main() {
  24.         fmt.Println(Add(2, 2))
  25. }
复制代码
这里遇到了一个Get “http://127.0.0.1:8000/add?a=1&b=2”: context deadline exceeded (Client.Timeout exceeded while awaiting headers)的问题，主要是客户端默认网络请求时间太短
修改之后
  1. package main
  3. import (
  4.         "encoding/json"
  5.         "fmt"
  6.         "time"
  8.         "github.com/kirinlabs/HttpRequest"
  9. )
  11. type ResponseData struct {
  12.         Data int `json:"data"`
  13. }
  15. func Add(a, b int) int {
  16.         req := HttpRequest.NewRequest()
  17.         req.SetTimeout(10 * time.Second)
  18.         res, _ := req.Get(fmt.Sprintf("http://127.0.0.1:8000/%s?a=%d&b=%d", "add", a, b))
  19.         body, _ := res.Body()
  20.         rspData := ResponseData{}
  21.         _ = json.Unmarshal(body, &rspData)
  22.         return rspData.Data
  23. }
  24. func main() {
  25.         fmt.Println(Add(2, 2))
  26. }
复制代码
rpc开发的要素分析

RPC技能在架构设计上有四部分构成，分别是：客户端、客户端存根、服务端、服务端存根。
客户端(Client)：服务调用发起方，也称为服务消费者。
客户端存根(Client Stub)：该步伐运行在客户端地点的计算机呆板上，主要用来存储要调用的服务器的地点，另外，该步伐还负责将客户端请求远端服务器步伐的数据信息打包成数据包，通过网络发送给服务端Stub步伐；其次，还要接收服务端Stub步伐发送的调用效果数据包，并剖析返回给客户端。
服务端(Server):远端的计算机呆板上运行的步伐，此中有客户端要调用的方法。
服务端存根(Server Stub)：接收客户Stub步伐通过网络发送的请求消息数据包，并调用服务端中真正的步伐功能方法，完乐成能调用；其次，将服务端实行调用的效果进行数据处理打包发送给客户端Stub步伐。
rpc需要使用到的术语

1、动态署理技能： 上文中我们提到的Client Stub和Sever Stub步伐，在具体的编码和开发实践过程中，都是使用动态署理技能自动生成的一段步伐。
序列化和反序列化： 在RPC调用的过程中，我们可以看到数据需要在一台呆板上传输到另外一台呆板上。在互联网上，所有的数据都是以字节的形式进行传输的。而我们在编程的过程中，每每都是使用数据对象，因此想要在网络上将数据对象和相关变量进行传输，就需要对数据对象做序列化和反序列化的操作。
序列化：把对象转换为字节序列的过程称为对象的序列化，也就是编码的过程。
反序列化：把字节序列规复为对象的过程称为对象的反序列化，也就是解码的过程。
我们常见的Json,XML等相关框架都可以对数据做序列化和反序列化编解码操作。背面我们要学习的Protobuf协议，这也是一种数据编解码的协议，在RPC框架中使用的更广泛。
简单的rpc实例

服务端
  1. package main
  3. import (
  4.         "net"
  5.         "net/rpc"
  6. )
  8. type HelloService struct {
  9. }
  11. func (s *HelloService) Hello(request string, reply *string) error {
  12.         //返回值是通过修改reply的值
  13.         *reply = "hello, " + request
  14.         return nil
  15. }
  16. func main() {
  17.         //rpc快速开发体验
  18.         //1.实例化一个server
  19.         listener, _ := net.Listen("tcp", ":1234")
  20.         //2.注册处理逻辑
  21.         _ = rpc.RegisterName("HelloService", &HelloService{})
  22.         //3.启动服务
  23.         conn, _ := listener.Accept() //当一个新的连接进来的时候
  24.         rpc.ServeConn(conn)
  25. }
复制代码
客户端
  1. package main
  3. import (
  4.         "encoding/json"
  5.         "fmt"
  6.         "net/rpc"
  7.         "time"
  8. )
  10. func main() {
  12.         //1.建立连接
  13.         client, err := rpc.Dial("tcp", "localhost:1234")
  14.         if err != nil {
  15.                 panic("连接失败")
  16.         }
  17.         var reply *string = new(string)
  18.         err = client.Call("HelloService.Hello", "bobby", reply)
  19.         if err != nil {
  20.                 panic("调用失败")
  21.         }
  22.         fmt.Println(*reply)
  23. }
复制代码
替换rpc的序列化协议为json

序列化协议为json，各种语言都可以调用服务端的内容
服务端
  1. package main
  3. import (
  4.         "net"
  5.         "net/rpc"
  6.         "net/rpc/jsonrpc"
  7. )
  9. type HelloService struct {
  10. }
  12. func (s *HelloService) Hello(request string, reply *string) error {
  13.         //返回值是通过修改reply的值
  14.         *reply = "hello, " + request
  15.         return nil
  16. }
  17. func main() {
  18.         //替换rpc的序列化协议为json
  19.         //1.实例化一个server
  20.         listener, _ := net.Listen("tcp", ":1234")
  21.         //2.注册处理逻辑
  22.         _ = rpc.RegisterName("HelloService", &HelloService{})
  23.         //3.启动服务
  24.         conn, _ := listener.Accept() //当一个新的连接进来的时候
  25.         rpc.ServeCodec(jsonrpc.NewServerCodec(conn))
  27. }
复制代码
客户端
  1. package main
  3. import (
  4.         "encoding/json"
  5.         "fmt"
  6.         "net"
  7.         "net/rpc"
  8.         "net/rpc/jsonrpc"
  9.         "time"
  11.         "github.com/kirinlabs/HttpRequest"
  12. )
  14. func main() {
  15.         //替换rpc的序列化协议为json
  16.         //1.建立连接
  17.         conn, err := net.Dial("tcp", "localhost:1234")
  18.         if err != nil {
  19.                 panic("连接失败")
  20.         }
  21.         var reply *string = new(string)
  22.         client := rpc.NewClientWithCodec(jsonrpc.NewClientCodec(conn))
  23.         err = client.Call("HelloService.Hello", "bobby", reply)
  24.         if err != nil {
  25.                 panic("调用失败")
  26.         }
  27.         fmt.Println(*reply)
  28. }
复制代码
python连接rpc，序列化协议为json

服务端
  1. package main
  3. import (
  4.         "net"
  5.         "net/rpc"
  6.         "net/rpc/jsonrpc"
  7. )
  9. type HelloService struct {
  10. }
  12. func (s *HelloService) Hello(request string, reply *string) error {
  13.         //返回值是通过修改reply的值
  14.         *reply = "hello, " + request
  15.         return nil
  16. }
  17. func main() {
  18.         //替换rpc的序列化协议为json
  19.         //1.实例化一个server
  20.         listener, _ := net.Listen("tcp", ":1234")
  21.         //2.注册处理逻辑
  22.         _ = rpc.RegisterName("HelloService", &HelloService{})
  23.         //3.启动服务
  24.         conn, _ := listener.Accept() //当一个新的连接进来的时候
  25.         rpc.ServeCodec(jsonrpc.NewServerCodec(conn))
  27. }
复制代码
客户端
  1. import json
  2. import socket
  3. request={
  4.     "id":0,
  5.     "params":["bobby"],
  6.     "method":"HelloService.Hello"
  7. }
  8. client=socket.create_connection(("localhost",1234))
  9. client.sendall(json.dumps(request).encode())
  10. #获取服务器返回的数据
  11. rsp=client.recv(4096)
  12. rsp=json.loads(rsp.decode())
  13. print(rsp)
复制代码
替换rpc的传输协议为http

服务端
  1. package main
  3. import (
  4.         "io"
  5.         "net/http"
  6.         "net/rpc"
  7.         "net/rpc/jsonrpc"
  8. )
  9. type HelloService struct {
  10. }
  12. func (s *HelloService) Hello(request string, reply *string) error {
  13.         //返回值是通过修改reply的值
  14.         *reply = "hello, " + request
  15.         return nil
  16. }
  17. func main() {
  18.         //替换rpc的序列化协议为http
  19.         //2.注册处理逻辑
  20.         _ = rpc.RegisterName("HelloService", &HelloService{})
  21.         http.HandleFunc("/jsonrpc", func(w http.ResponseWriter, r *http.Request) {
  22.                 var conn io.ReadWriteCloser = struct {
  23.                         io.Writer
  24.                         io.ReadCloser
  25.                 }{
  26.                         ReadCloser: r.Body,
  27.                         Writer:     w,
  28.                 }
  29.                 rpc.ServeRequest(jsonrpc.NewServerCodec(conn))
  30.         })
  31.         http.ListenAndServe(":1234", nil)
  32. }
复制代码
客户端
  1. import requests
  2. request={
  3.     "id":0,
  4.     "params":["bobby"],
  5.     "method":"HelloService.Hello"
  6. }
  7. rsp=requests.post("http://localhost:1234/jsonrpc",json=request)
  8. print(rsp.text)
复制代码
客户端
  1. package main
  3. import (
  4.         "encoding/json"
  5.         "fmt"
  6.         "net/rpc"
  7.         "time"
  9.         "github.com/kirinlabs/HttpRequest"
  10. )
  12. func main() {
  13.         //1.实例化一个
  14.         client, err := rpc.Dial("tcp", "127.0.0.1:1234")
  15.         if err != nil {
  16.                 panic("连接失败")
  17.         }
  18.         var reply string
  19.         err = client.Call("HelloService.Hello", "bobby", &reply)
  20.         if err != nil {
  21.                 // panic("调用失败")
  22.                 fmt.Println(err)
  23.         }
  24.         fmt.Println(reply)
  25. }
复制代码
署理封装

署理类
  1. package handler
  3. import "net/rpc"
  5. const HelloServiceName = "HelloServiceName"
  7. type HelloServiceStub struct {
  8.         *rpc.Client
  9. }
  11. func NewHelloServiceClient(protcol, address string) HelloServiceStub {
  12.         conn, err := rpc.Dial(protcol, address)
  13.         if err != nil {
  14.                 panic("connect error!")
  15.         }
  16.         return HelloServiceStub{conn}
  17. }
  18. func (c *HelloServiceStub) Hello(request string, reply *string) error {
  19.         err := c.Call(HelloServiceName+".Hello", request, reply)
  20.         if err != nil {
  21.                 return err
  22.         }
  23.         return nil
  24. }
  26. type HelloService struct {
  27. }
  29. func (s *HelloService) Hello(request string, reply *string) error {
  30.         //返回值是通过修改reply的值
  31.         *reply = "hello, " + request
  32.         return nil
  33. }
  35. func RegisterHelloService() error {
  36.         return rpc.RegisterName(HelloServiceName, &HelloService{})
  37. }
复制代码
服务端
  1. package main
  3. import (
  4.         "TEMP/handler"
  5.         "net"
  6.         "net/rpc"
  7. )
  9. func main() {
  10.         //进一步改造rpc调用的代码
  11.         //1.实例化一个server
  12.         listener, _ := net.Listen("tcp", ":1234")
  13.         //2.注册处理逻辑handler
  14.         _ = handler.RegisterHelloService()
  15.         //3.启动服务
  16.         conn, _ := listener.Accept() //当一个新的连接进来的时候
  17.         rpc.ServeConn(conn)
  19. }
复制代码
客户端
  1. package main
  3. import (
  4.         "TEMP/handler"
  5.         "fmt"
  6. )
  8. func main() {
  9.         //进一步改造rpc调用的代码
  10.         //1.实例化一个
  11.         client := handler.NewHelloServiceClient("tcp", "127.0.0.1:1234")
  12.         var reply string
  13.         err := client.Hello("bobby", &reply)
  14.         if err != nil {
  15.                 // panic("调用失败")
  16.                 fmt.Println(err)
  17.         }
  18.         fmt.Println(reply)
  19. }
复制代码
解耦合

将服务端的Hello函数传递变换成接口类型
署理类
  1. package handler
  3. import "net/rpc"
  5. const HelloServiceName = "HelloServiceName"
  7. type HelloServiceStub struct {
  8.         *rpc.Client
  9. }
  11. func NewHelloServiceClient(protcol, address string) HelloServiceStub {
  12.         conn, err := rpc.Dial(protcol, address)
  13.         if err != nil {
  14.                 panic("connect error!")
  15.         }
  16.         return HelloServiceStub{conn}
  17. }
  18. func (c *HelloServiceStub) Hello(request string, reply *string) error {
  19.         err := c.Call(HelloServiceName+".Hello", request, reply)
  20.         if err != nil {
  21.                 return err
  22.         }
  23.         return nil
  24. }
  26. type NewHelloService struct {
  27. }
  28. type HelloServicer interface {
  29.         Hello(request string, reply *string) error
  30. }
  32. func (s *NewHelloService) Hello(request string, reply *string) error {
  33.         //返回值是通过修改reply的值
  34.         *reply = "hello, " + request
  35.         return nil
  36. }
  38. func RegisterHelloService() error {
  39.         return rpc.RegisterName(HelloServiceName, &NewHelloService{})
  40. }
复制代码
服务端
  1. package main
  3. import (
  4.         "TEMP/handler"
  5.         "net"
  6.         "net/rpc"
  7. )
  9. func main() {
  10.         //进一步改造rpc调用的代码
  11.         //1.实例化一个server
  12.         listener, _ := net.Listen("tcp", ":1234")
  13.         //2.注册处理逻辑handler
  14.         _ = handler.RegisterHelloService()
  15.         //3.启动服务
  16.         conn, _ := listener.Accept() //当一个新的连接进来的时候
  17.         rpc.ServeConn(conn)
  19. }
复制代码
客户端
  1. package main
  3. import (
  4.         "TEMP/handler"
  5.         "fmt"
  6. )
  8. func main() {
  9.         //进一步改造rpc调用的代码
  10.         //1.实例化一个
  11.         client := handler.NewHelloServiceClient("tcp", "127.0.0.1:1234")
  12.         var reply string
  13.         err := client.Hello("bobby", &reply)
  14.         if err != nil {
  15.                 // panic("调用失败")
  16.                 fmt.Println(err)
  17.         }
  18.         fmt.Println(reply)
  19. }
复制代码
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。

本帖子中包含更多资源

您需要 登录 才可以下载或查看，没有账号？立即注册

x
回复

使用道具 举报

0 个回复

倒序浏览
返回列表

快速回复

高级模式
B Color Image Link Quote Code Smilies
您需要登录后才可以回帖 登录 or 立即注册

本版积分规则

东湖之滨

金牌会员
这个人很懒什么都没写!

楼主热帖

标签云

挺好的 服务器
微信订阅号
微信服务号
微信客服
小程序
H5
关于我们 商务合作 网站地图
©Copyright 2022-2024 杭州祥升科技有限公司 版权所有 浙ICP备20004199号
快速回复 返回顶部 返回列表