HarmonyOS开发者限时福利来啦!最高10w+现金激励等你拿~ 了解详情
写点什么

300 行 Go 代码玩转 RPC

  • 2019-11-14
  • 本文字数:4311 字

    阅读完需:约 14 分钟

300行Go代码玩转RPC

最近,小编一直在研究 RPC 的原理及实现方式。在本篇文章中将通过用 300 行纯 Golang 编写简单的 RPC 框架来解释 RPC。希望能帮助大家梳理 RPC 相关知识点。


我们通过从头开始在 Golang 中构建一个简单的 RPC 框架来学习 RPC 基础构成。

1 什么是 RPC

简单地说,服务 A 想调用服务 B 的函数。但是这两个服务不在同一个内存空间中。所以不能直接调用它。


因此,为了实现这个调用,我们需要表达如何调用以及如何通过网络传递通信的语义。


让我们考虑一下,当我们可以在相同的内存空间(本地调用)中运行时,我们要怎么做。


type User struct {  Name string  Age int}
var userDB = map[int]User{ 1: User{"Ankur", 85}, 9: User{"Anand", 25}, 8: User{"Ankur Anand", 27},}

func QueryUser(id int) (User, error) { if u, ok := userDB[id]; ok { return u, nil }
return User{}, fmt.Errorf("id %d not in user db", id)}

func main() { u , err := QueryUser(8) if err != nil { fmt.Println(err) return }
fmt.Printf("name: %s, age: %d \n", u.Name, u.Age)}
复制代码


现在我们如何在网络上进行相同的函数调用


客户端将通过网络调用 QueryUser(id int) 函数,并且将有一个服务端提供对该函数的调用,并返回响应 User{“Name”, id}, nil。

2 网络传输数据格式

我们将采用 TLV(定长报头+变长消息体)编码方案来规范 tcp 上的数据传输。稍后会详细介绍


在通过网络发送数据之前,我们需要定义如何通过网络发送数据的结构。


这有助于我们定义一个通用协议,客户端和服务端都可以理解这个协议。(protobuf IDL 定义了服务端和客户端都能理解的内容)。


因此,服务端接收到的数据、要调用的函数名和参数列表,或者来自客户端的数据都需要传递这些参数。


另外,让我们约定第二个返回值的类型为 error,表示 RPC 调用结果。


// RPC数据传输格式type RPCdata struct {  Name string        // name of the function  Args []interface{} // request's or response's body expect error.  Err  string        // Error any executing remote server}
复制代码


现在我们有了一个格式,我们需要序列化它以便我们可以通过网络发送它。在本例中,我们将使用 go 默认的二进制序列化协议进行编码和解码。


// be sent over the network.func Encode(data RPCdata) ([]byte, error) {  var buf bytes.Buffer  encoder := gob.NewEncoder(&buf)  if err := encoder.Encode(data); err != nil {    return nil, err  }  return buf.Bytes(), nil}
// Decode the binary data into the Go structfunc Decode(b []byte) (RPCdata, error) { buf := bytes.NewBuffer(b) decoder := gob.NewDecoder(buf) var data RPCdata if err := decoder.Decode(&data); err != nil { return Data{}, err } return data, nil}
复制代码

3 网络传输

选择 TLV 协议的原因是由于其非常容易实现,同时也完成了我们需要识别的数据读取的长度,因为我们需要确定这个请求读取的字节数的传入请求流。发送和接收都执行相同的操作。


// Transport will use TLV protocoltype Transport struct {  conn net.Conn // Conn is a generic stream-oriented network connection.}
// NewTransport creates a Transportfunc NewTransport(conn net.Conn) *Transport { return &Transport{conn}}
// Send TLV data over the networkfunc (t *Transport) Send(data []byte) error { // we will need 4 more byte then the len of data // as TLV header is 4bytes and in this header // we will encode how much byte of data // we are sending for this request. buf := make([]byte, 4+len(data)) binary.BigEndian.PutUint32(buf[:4], uint32(len(data))) copy(buf[4:], data) _, err := t.conn.Write(buf) if err != nil { return err } return nil}
// Read TLV sent over the wirefunc (t *Transport) Read() ([]byte, error) { header := make([]byte, 4) _, err := io.ReadFull(t.conn, header) if err != nil { return nil, err } dataLen := binary.BigEndian.Uint32(header) data := make([]byte, dataLen) _, err = io.ReadFull(t.conn, data) if err != nil { return nil, err } return data, nil}
复制代码


现在我们已经定义了数据格式和传输协议。下面我们还需要 RPC 服务器和 RPC 客户端的实现。

4 RPC 服务器

RPC 服务器将接收具有函数名的 RPCData。因此,我们需要维护和映射包含函数名到实际函数映射的函数


// RPCServer ...type RPCServer struct {  addr string  funcs map[string] reflect.Value}
// Register the name of the function and its entriesfunc (s *RPCServer) Register(fnName string, fFunc interface{}) { if _,ok := s.funcs[fnName]; ok { return }
s.funcs[fnName] = reflect.ValueOf(fFunc)}
复制代码


现在我们已经注册了 func,当我们收到请求时,我们将检查函数执行期间传递的 func 的名称是否存在。然后执行相应的操作


// Execute the given function if presentfunc (s *RPCServer) Execute(req RPCdata) RPCdata {  // get method by name  f, ok := s.funcs[req.Name]  if !ok {    // since method is not present    e := fmt.Sprintf("func %s not Registered", req.Name)    log.Println(e)    return RPCdata{Name: req.Name, Args: nil, Err: e}  }
log.Printf("func %s is called\n", req.Name) // unpackage request arguments inArgs := make([]reflect.Value, len(req.Args)) for i := range req.Args { inArgs[i] = reflect.ValueOf(req.Args[i]) }
// invoke requested method out := f.Call(inArgs) // now since we have followed the function signature style where last argument will be an error // so we will pack the response arguments expect error. resArgs := make([]interface{}, len(out) - 1) for i := 0; i < len(out) - 1; i ++ { // Interface returns the constant value stored in v as an interface{}. resArgs[i] = out[i].Interface() }
// pack error argument var er string if e, ok := out[len(out) - 1].Interface().(error); ok { // convert the error into error string value er = e.Error() } return RPCdata{Name: req.Name, Args: resArgs, Err: er}}
复制代码

5 RPC 客户端

由于函数的具体实现在服务器端,客户端只有函数的原型,所以我们需要调用函数的完整原型,这样我们才能调用它。


func (c *Client) callRPC(rpcName string, fPtr interface{}) {  container := reflect.ValueOf(fPtr).Elem()  f := func(req []reflect.Value) []reflect.Value {    cReqTransport := NewTransport(c.conn)    errorHandler := func(err error) []reflect.Value {      outArgs := make([]reflect.Value, container.Type().NumOut())      for i := 0; i < len(outArgs)-1; i++ {        outArgs[i] = reflect.Zero(container.Type().Out(i))      }      outArgs[len(outArgs)-1] = reflect.ValueOf(&err).Elem()      return outArgs    }
// Process input parameters inArgs := make([]interface{}, 0, len(req)) for _, arg := range req { inArgs = append(inArgs, arg.Interface()) }
// ReqRPC reqRPC := RPCdata{Name: rpcName, Args: inArgs} b, err := Encode(reqRPC) if err != nil { panic(err) } err = cReqTransport.Send(b) if err != nil { return errorHandler(err) } // receive response from server rsp, err := cReqTransport.Read() if err != nil { // local network error or decode error return errorHandler(err) } rspDecode, _ := Decode(rsp) if rspDecode.Err != "" { // remote server error return errorHandler(errors.New(rspDecode.Err)) }
if len(rspDecode.Args) == 0 { rspDecode.Args = make([]interface{}, container.Type().NumOut()) } // unpackage response arguments numOut := container.Type().NumOut() outArgs := make([]reflect.Value, numOut) for i := 0; i < numOut; i++ { if i != numOut-1 { // unpackage arguments (except error) if rspDecode.Args[i] == nil { // if argument is nil (gob will ignore "Zero" in transmission), set "Zero" value outArgs[i] = reflect.Zero(container.Type().Out(i)) } else { outArgs[i] = reflect.ValueOf(rspDecode.Args[i]) } } else { // unpackage error argument outArgs[i] = reflect.Zero(container.Type().Out(i)) } }
return outArgs } container.Set(reflect.MakeFunc(container.Type(), f))}
复制代码

6 测试一下我们的框架

package main
import ( "encoding/gob" "fmt" "net")
type User struct { Name string Age int}
var userDB = map[int]User{ 1: User{"Ankur", 85}, 9: User{"Anand", 25}, 8: User{"Ankur Anand", 27},}
func QueryUser(id int) (User, error) { if u, ok := userDB[id]; ok { return u, nil }
return User{}, fmt.Errorf("id %d not in user db", id)}
func main() { // new Type needs to be registered gob.Register(User{}) addr := "localhost:3212" srv := NewServer(addr)
// start server srv.Register("QueryUser", QueryUser) go srv.Run()
// wait for server to start. time.Sleep(1 * time.Second)
// start client conn, err := net.Dial("tcp", addr) if err != nil { panic(err) } cli := NewClient(conn)
var Query func(int) (User, error) cli.callRPC("QueryUser", &Query)
u, err := Query(1) if err != nil { panic(err) } fmt.Println(u)
u2, err := Query(8) if err != nil { panic(err) } fmt.Println(u2)}
复制代码


执行:go run main.go


输出内容


2019/07/23 20:26:18 func QueryUser is called{Ankur 85}2019/07/23 20:26:18 func QueryUser is called{Ankur Anand 27}
复制代码

总结

致此我们简单的 RPC 框架就实现完成了,旨在帮大家理解 RPC 的原理及上手简单实践。如果大家对这篇文章中所讲内容有异议,或者想进一步讨论,请留言回复。


本文转载自公众号 360 云计算(ID:hulktalk)。


原文链接:


https://mp.weixin.qq.com/s/fxrocOMLX7kqUH9lP94JpA


2019-11-14 17:341268

评论 1 条评论

发布
用户头像
请问有源码吗,少了Run方法
2020-07-28 11:14
回复
没有更多了
发现更多内容

​​苹果应用上架后多久可以下载​

雪奈椰子

ios打包

Spring Cloud Alibaba x AppActive 带来的全新异地活动解决方案

阿里巴巴中间件

阿里云 云原生 Spring Cloud Aliababa

Guitar Pro2024免费版吉他打谱软件

茶色酒

Guitar Pro8 Guitar Pro2024

​​软件开发入门教程网之​​C++ 标准库

雪奈椰子

C++

到底什么样的 REST 才是最佳 REST?

江南一点雨

Rest springboot

大数据技术培训学习机构怎么样

小谷哥

程序摄像头Trace Profiling:生产环境10分钟黄金时间快速排障手册

KINDLING

Java Linux 运维 ebpf

openEuler委员会主席江大勇:激发原创力量,逐梦数智未来

openEuler

数据库 开源 操作系统 openEuler 资讯

对话阿里云叔同:如何看待 2022 年云原生的发展,2023 年有哪些值得关注的技术?

Serverless Devs

阿里云 Serverless 云原生

一站式开发平台 赋能办公全场景

力软低代码开发平台

学习大数据技术培训班靠谱吗

小谷哥

java程序员培训学习值得吗?

小谷哥

对话开发者:Serverless 落地的困境与破局

Serverless Devs

阿里云 Serverless 云原生

聊聊降本提效这件事儿

阿里巴巴中间件

阿里云 容器 云原生 中间件 降本增效

容器 I/O 性能诊断:到底哪个应用是带宽杀手?

阿里巴巴中间件

阿里云 容器 云原生

武汉web前端培训靠谱吗

小谷哥

WEB前端编程培训学技术怎么样

小谷哥

2023 好运开年,OpenMLDB 入选 2022 中国技术品牌影响力企业

第四范式开发者社区

人工智能 机器学习 数据库 开源 特征

比 CK 再快 24% ! YMatrix 5.0 SSB 基准测试报告

YMatrix 超融合数据库

Clickhouse 性能基准测试 超融合数据库 YMatrix

从0.5到4.0,OceanBase单机分布式一体化的技术演进|DTCC 2022

OceanBase 数据库

数据库 oceanbase

​​软件开发入门教程网之​​C++ 信号处理信号

雪奈椰子

OceanBase 4.0解读:从TPC-H性能测评看4.0与3.x差异

OceanBase 数据库

数据库 oceanbase

深耕5G云专网,阿里云祝顺民入选“2022年度5G创新人物”

云布道师

云网络

餐饮业数字化提速,OceanBase助海底捞变身“云上捞”

OceanBase 数据库

数据库 oceanbase

【Meetup预告】SeaTunnel + OpenMLDB:共筑数据集成生态,加速实时场景落地

第四范式开发者社区

人工智能 机器学习 数据库 开源 特征

有什么好用的云渲染?这篇文章给你答案

Renderbus瑞云渲染农场

云渲染 好用的云渲染平台

苹果开发者账号可以多人使用吗​​​

雪奈椰子

ios打包

微软提出通用解码器 X-Decoder,支持图像分割和语言分词

Zilliz

算法模型

重磅 | 九科信息加入深圳市智能制造产业促进会,共促智能制造产业健康发展

九科Ninetech

数智化转型进入“精装时代”,容联云助力千行百业加速上云用数赋智

脑极体

300行Go代码玩转RPC_文化 & 方法_360云计算_InfoQ精选文章