当前位置： 首页 > news >正文

【GeeRPC】Day1：服务端与消息编码

news 2025/2/8 11:15:16

勘误：解决了第一次编写时候产生的 bug

第一次完成简易的客户端的时候，产生了 bug：

rpc server: read argv err: gob: decoding into local type *string, received remote type Header = struct { ServiceMethod string; Seq uint; Error string; }

bug 产生的原因是 codec/gob.go 当中有一个参数填写错误了，GobCodec 的 Write 方法当中：

func (c *GobCodec) Write(h *Header, body interface{}) (err error) {defer func() {_ = c.buf.Flush()if err != nil {_ = c.Close()}}()if err = c.enc.Encode(h); err != nil {log.Println("rpc: gob error encoding header:", err)return}if err = c.enc.Encode(body); err != nil {		// 此处在第一次编写的时候出错了, 应该是 body，错填为 hlog.Println("rpc: gob error encoding body:", err)return}return
}

Day1：服务端与消息编码

今日任务：

  • 使用 encoding/gob 实现消息的编解码（序列化和反序列化，它是 RPC 框架需要解决的一个难题之一）；
  • 实现一个简易的服务端，仅接收消息，不处理，代码约 200 行。（客户端向服务端发送消息，服务端处理消息将结果反馈给客户端，这就是 RPC 的基本流程）

目前目录的组织形式如下：
在这里插入图片描述

消息的序列化和反序列化

一个典型的 RPC 调用如下：

err := client.Call("Arith.Multiply", args, &reply)
// 推测是使用字符串描述远程调用的函数名, args 应该是一个保存参数的序列, reply 保存计算结果, 传址调用.

Arith是服务名，Multiply是具体的方法名，参数为args，服务端的响应包括错误error和返回值reply。

将请求和响应中的参数和返回值抽象为 body，剩余的信息放在 header 中，可以抽象出数据结构 Header：

// in /geerpc/codec/codec.go
package codectype Header struct {ServiceMethod string // format: "Service.Method"Seq           uint64 // sequence number chosen by clientError         string
}
  • ServiceMethod 是服务名和方法名，常与 Golang 中的结构体和方法相映射；
  • Seq 是请求的序列号，可以认为是某个请求的 ID，用来区分不同的请求；
  • Error 是错误信息，客户端置为空，服务端如果发生错误，将错误信息置于 Error 当中；

进一步抽象出对消息体进行编解码的接口 Codec，抽象出接口是为了实现不同的 Codec 实例【在后面我们将会看到，分别定义了 JSON 和 Gob 两种 Codec 实例，但这个教程仅实现了在 Gob 上的方法】：

type Codec interface {io.CloserReadHeader(*Header)	errorReadBody(interface{}) errorWrite(*Header, interface{}) error
}

紧接着抽象出 Codec 的构造函数，客户端和服务端可以通过 Codec 的 Type 得到构造函数，从而创建 Codec 实例。这部分代码和工厂模式【来自 DeepSeek：工厂模式（Factory Pattern）是一种创建型设计模式，用于创建对象而不指定具体的类。它通过定义一个接口或抽象类来创建对象，并由子类决定实例化哪个类】类似，与工厂模式不同的是，返回的是构造函数而非实例【由构造函数进一步返回实例】：

type NewCodecFunc func(closer io.ReadWriteCloser) Codectype Type stringconst (GobType  Type = "application/gob"JsonType Type = "application/json"
)var NewCodecFuncMap map[Type]NewCodecFuncfunc init() {NewCodecFuncMap = make(map[Type]NewCodecFunc)	// 仅实现了 GobNewCodecFuncMap[GobType] = NewGobCodec			// NewGobCodec 还没定义, 将在 gob.go 定义
}

我们定义了两种 Codec，即Gob和JSON，但实际代码中只实现了Gob一种。事实上，二者的实现非常接近，只需要把gob换为json即可。

首先定义GobCodec结构体，它由四部分构成，conn是由构建函数传入，通常是通过 TCP 或 Unix 建立 socket 时得到的链接实例，dec 和 enc 对应 gob 的 Decoder 和 Encoder，buf 是为了防止阻塞而创建的带缓冲的Writer，一般这么做是为了提升性能：

// in geerpc/codec/gob.go
package codecimport ("bufio""encoding/gob""io"
)type GobCodec struct {			// 专门解码 Gob 的 encoder-decoderconn io.ReadWriteCloserbuf  *bufio.Writerdec  *gob.Decoderenc  *gob.Encoder
}var _ Codec = (*GobCodec)(nil)func NewGobCodec(conn io.ReadWriteCloser) Codec {	// GobCodec 的工厂函数buf := bufio.NewWriter(conn)return &GobCodec{								// 与 C++ 非常不同的是, golang 可以返回局部变量conn: conn,buf:  buf,dec:  gob.NewDecoder(conn),enc:  gob.NewEncoder(buf),}
}

【此时 NewGobCodec 会标红，因为这个函数的返回值是 Codec 接口，还应该实现 GobCodec 的 Close、Write、ReadHeader、ReadBody 等方法】
进一步实现接口的方法：

func (c *GobCodec) Close() error {	// 关闭连接return c.conn.Close()
}func (c *GobCodec) ReadHeader(h *Header) error {	// 解码 Headerreturn c.dec.Decode(h)
}func (c *GobCodec) ReadBody(body interface{}) error {	// 解码 Bodyreturn c.dec.Decode(body)
}func (c *GobCodec) Write(h *Header, body interface{}) error {	// 写入defer func() {err := c.buf.Flush()if err != nil {_ = c.Close()}}()if err := c.enc.Encode(h); err != nil {log.Println("rpc codec: gob error encoding header:", err)return err}if err := c.enc.Encode(body); err != nil {log.Println("rpc codec: gob error encoding body:", err)return err}return nil
}

通信过程

客户端和服务端的通信需要协商一些内容，比如 HTTP 报文，分为 header 和 body 两部分【因此 Codec 接口需要实现 ReadBody 和 ReadHeader 两个方法】，body 的格式和长度通过 header 中的content-type和content-length来指定，服务端通过解析 header 就能够知道如何从 body 中读取需要的信息。对于 RPC 协议来说，这部分协商是需要自主设计的。为了提升性能，一般在报文最开始会规划固定的字节，来协商相关信息。比如第一个字节用来表示序列化方式，第二个字节表示压缩方式，第三到六个字节表示 header 的长度，七到十表示 body 的长度。

对 GeeRPC 来说，目前唯一需要协商的是消息的解码方式。我们将这部分信息放到结构体Option当中承载。目前，已经进入服务端的实现阶段了：

// in geerpc/server.go
package geerpcimport "Geektutu/GeeRPC/geerpc/codec"const MagicNumber = 0x3bef5ctype Option struct {MagicNumber intCodecType   codec.Type
}var DefaultOption = &Option {MagicNumber: MagicNumber,CodecType: codec.GobType,
}

一般来说，涉及协议协商这部分的信息，需要设计固定的字节来传输。但是为了实现上的简单，GeeRPC 客户端固定采用 JSON 编码 Option，后续的 header 和 body 的编码方式由 Option 中的 CodeType 来指定，服务端首先使用 JSON 解码 Option，然后通过 Option 的 CodeType 解码剩余的内容。也就是说，报文将会以下述形式发送：

| Option{MagicNumber: xxx, CodecType: xxx} | Header{ServiceMethod ...} | Body interface{} |
| <------      固定 JSON 编码      ------>  | <-------   编码方式由 CodeType 决定   ------->|

在一次连接中，Option 固定在报文的最开始，Header 和 Body 可以有多个，即报文可能是这样的：

| Option | Header1 | Body1 | Header2 | Body2 | ...

服务端的实现

通信过程定义清楚之后，服务端的实现就比较直接了：

// still in geerpc/server.go
// Server represents a RPC Server.
type Server struct{}// NewServer returns a new Server.
func NewServer() *Server {return &Server{}
}// DefaultServer is the default instance of Server
var DefaultServer = NewServer()// Accept accepts connections on the listener and server
// requests for each incoming connection
func (server *Server) Accept(lis net.Listener) {for {conn, err := lis.Accept()if err != nil {log.Println("rpc server: accept error:", err)return}go server.ServeConn(conn)}
}// Accept accepts connections on the listener and serves requests for each incoming connection
func Accept(lis net.Listener) { DefaultServer.Accept(lis) }
  • 首先定义了结构体Server，它没有任何成员字段。
  • 实现了Accept方法，net.Listener作为参数，for 循环等待 socket 连接建立，并开启子协程处理（通过 go 关键字来完成），处理过程交给了Server的ServeConn方法，将在下面进行实现。
  • DefaultServer 是一个默认的 Server 实例，方便用户使用。

如果想启动服务，过程非常简单，传入 listener 即可：

lis, _ := net.Listen("tcp", ":9999")
geerpc.Accept(lis)

ServeConn的实现就和之前讨论的通信过程关系非常紧密了，首先使用json.NewDecoder反序列化得到 Option 实例，检查 MagicNumber 和 CodeType 的值是否正确。然后根据 CodeType 得到对应的消息编解码器，接下来的处理交给serverCodec。

// ServeConn runs the server on a single connection.
// ServeConn blocks, serving the connection until the client hangs up.
func (server *Server) ServeConn(conn io.ReadWriteCloser) {defer func() { _ = conn.Close() }()var opt Optionif err := json.NewDecoder(conn).Decode(&opt); err != nil {log.Println("rpc server: options error: ", err)return}if opt.MagicNumber != MagicNumber {log.Printf("rpc server: invalid magic number %x", opt.MagicNumber)return}f := codec.NewCodecFuncMap[opt.CodecType]if f == nil {log.Printf("rpc server: invalid codec type %s", opt.CodecType)return}server.serveCodec(f(conn))
}// invalidRequest is a placeholder for response argv when error occurs
var invalidRequest = struct{}{}func (server *Server) serveCodec(cc codec.Codec) {sending := new(sync.Mutex) // make sure to send a complete responsewg := new(sync.WaitGroup)  // wait until all requests are handledfor {req, err := server.readRequest(cc)if err != nil {if req == nil {break // it's not possible to recover, so close the connection}req.h.Error = err.Error()server.sendResponse(cc, req.h, invalidRequest, sending)continue}wg.Add(1)go server.handleRequest(cc, req, sending, wg)}wg.Wait()_ = cc.Close()
}

serveCodec的过程非常简单，主要包含三个阶段：

  • 读取请求：readRequest；
  • 处理请求：handleRequest；
  • 回复请求：sendResponse；

之前提到过，在一次连接中，允许接收多个请求，即多个 request header 和 request body，因此在 serveCodec 中使用 for 永真循环等待请求的到来，直到发生错误（比如连接被关闭，接收到的报文有问题等），这里需要注意三个点：

  • handleRequest 使用协程并发处理执行请求；
  • 处理请求是并发的，但是回复请求的报文必须逐个发送，并发容易导致多个回复报文交织在一起，客户端无法解析。在此使用锁（sending）来保证；
  • 尽力而为，只有在 header 解析失败时，才终止循环。

下面要做的是补齐 server 对象缺失的方法：

type request struct {h            *codec.Headerargv, replyv reflect.Value
}func (server *Server) readRequestHeader(cc codec.Codec) (*codec.Header, error) {var h codec.Headerif err := cc.ReadHeader(&h); err != nil {if err != io.EOF && err != io.ErrUnexpectedEOF {log.Println("rpc server: read header error:", err)}return nil, err}return &h, nil
}func (server *Server) readRequest(cc codec.Codec) (*request, error) {h, err := server.readRequestHeader(cc)if err != nil {return nil, err}req := &request{h: h}// TODO: now we don't know the type of request argv// day 1, just suppose it's stringreq.argv = reflect.New(reflect.TypeOf(""))if err = cc.ReadBody(req.argv.Interface()); err != nil {log.Println("rpc server: read argv err:", err)}return req, nil
}func (server *Server) sendResponse(cc codec.Codec, h *codec.Header, body interface{}, sending *sync.Mutex) {sending.Lock()defer sending.Unlock()if err := cc.Write(h, body); err != nil {log.Println("rpc server: write response errro:", err)}
}func (server *Server) handleRequest(cc codec.Codec, req *request, sending *sync.Mutex, wg *sync.WaitGroup) {// TODO, should call registered rpc methods to get the right replyv// day 1, just print argv and send a hello messagedefer wg.Done()log.Println(req.h, req.argv.Elem())req.replyv = reflect.ValueOf(fmt.Sprintf("geerpc resp %d", req.h.Seq))server.sendResponse(cc, req.h, req.replyv.Interface(), sending)
}

目前尚不能判断 body 的类型，因此在 readRequest 和 handleRequest 中，我们在第一天假定把 body 当作字符串处理。接收到请求之后，打印 header，并回复geerpc resp ${req.h.Seq}。

main 函数（一个简易的客户端）

第一天的内容已经接近尾声（说实话，有些难懂），我们已经实现了一个消息的编解码器GobCodec，并且客户端与服务端实现了简单的协议交换（protocol exchange），即允许客户端使用不同的编码方式。同时实现了服务器的雏形，可以建立连接、读取、处理并回复客户端的请求。

接下来建立一个 main 函数来看一下如何使用刚刚实现的 GeeRPC：

package mainimport ("Geektutu/GeeRPC/geerpc""Geektutu/GeeRPC/geerpc/codec""encoding/json""fmt""log""net""time"
)func startServer(addr chan string) {l, err := net.Listen("tcp", ":0")if err != nil {log.Fatal("network error:", err)}log.Println("start rpc server on", l.Addr())addr <- l.Addr().String()geerpc.Accept(l)
}func main() {addr := make(chan string)go startServer(addr)conn, _ := net.Dial("tcp", <-addr)defer func() { _ = conn.Close() }()time.Sleep(time.Second)// send options_ = json.NewEncoder(conn).Encode(geerpc.DefaultOption)cc := codec.NewGobCodec(conn)// send request & receive responsefor i := 0; i < 5; i++ {h := &codec.Header{ServiceMethod: "Foo.Sum",Seq:           uint64(i),}_ = cc.Write(h, fmt.Sprintf("geerpc req %d", h.Seq))_ = cc.ReadHeader(h)var reply string_ = cc.ReadBody(&reply)log.Println("reply:", reply)}
}

目前直接运行这一段代码会在：

func (server *Server) readRequest(cc codec.Codec) (*request, error) {h, err := server.readRequestHeader(cc)if err != nil {return nil, err}req := &request{h: h}// TODO: now we don't know the type of request argv// day 1, just suppose it's stringreq.argv = reflect.New(reflect.TypeOf(""))if err = cc.ReadBody(req.argv.Interface()); err != nil {log.Println("rpc server: read argv err:", err)}return req, nil
}

这一片段的 if err = cc.ReadBody(req.argv.Interface()); err != nil 报错，报错信息如下：

rpc server: read argv err: gob: decoding into local type *string, received remote type Header = struct { ServiceMethod string; Seq uint; Error string; }

应该是解码上的问题，目前还没有找到解决方法，希望这部分会在后续的学习过程中解决。

**最新：**这部分 bug 已经解决，详见文章开头的勘误。


http://www.mrgr.cn/news/89799.html

相关文章：

  • 数据结构 图
  • 如何将本地 Node.js 服务部署到宝塔面板：完整的部署指南
  • LabVIEW与WPS文件格式的兼容性
  • pytest-allure框架简单封装----测试报告
  • 认识机器学习中的经验风险最小化准则
  • IaaS、PaaS、SaaS 和 FaaS
  • 网络编程 day2
  • .net8.0使用EF连接sqlite数据库及使用Gridify实现查询的简易实现
  • 2025.2.7 Python开发岗面试复盘
  • 一文吃透！DataStage 全面概述与核心知识要点大公开
  • 如何在Windows上使用Docker
  • xinference 安装（http导致错误解决）
  • hive的几种复杂数据类型
  • 深度学习01 神经网络
  • 使用bucardo实现postgresql数据库双主同步
  • 一文速览DeepSeek-R1的本地部署——可联网、可实现本地知识库问答：包括671B满血版和各个蒸馏版的部署
  • 二分查找算法 （典型算法思想）—— OJ例题算法解析思路
  • MFC 学习笔记目录
  • 车型检测7种YOLOV8
  • 订单状态监控实战：基于 SQL 的状态机分析与异常检测
  • 制造业设备状态监控与生产优化实战：基于SQL的序列分析与状态机建模
  • Denavit-Hartenberg DH MDH坐标系
  • 深入解析 COUNT(DISTINCT) OVER(ORDER BY)：原理、问题与高效替代方案
  • 芯片AI深度实战：让verilog不再是 AI 的小众语言
  • SQL进阶实战技巧：某芯片工厂设备任务排产调度分析 | 间隙分析技术应用
  • android 音视频系列引导