day1 服务端与消息编码

本文代码地址：
本文是7天用Go从零实现RPC框架GeeRPC的第一篇。

• 使用 encoding/gob 实现消息的编解码(序列化与反序列化)
  
• 实现一个浅易的服务端，仅接受消息，不处理，代码约 200 行
  

消息的序列化与反序列化

一个典型的 RPC 调用如下：

1. err = client.Call("Arith.Multiply", args, &reply)

复制代码

客户端发送的哀求包括服务名 Arith，方法名 Multiply，参数 args 三个，服务端的响应包括错误 error，返回值 reply 2 个。我们将哀求和响应中的参数和返回值抽象为 body，剩余的信息放在 header 中，那么就可以抽象出数据布局 Header：
day1-codec/codec/codec.go

1. package codec
3. import "io"
5. type Header struct {
6.         ServiceMethod string // format "Service.Method"
7.         Seq           uint64 // sequence number chosen by client
8.         Error         string
9. }

复制代码

• ServiceMethod 是服务名和方法名，通常与 Go 语言中的布局体和方法相映射。
  
• Seq 是哀求的序号，也可以认为是某个哀求的 ID，用来区分不同的哀求。
  
• Error 是错误信息，客户端置为空，服务端如果如果发生错误，将错误信息置于 Error 中。
  

我们将和消息编解码相关的代码都放到 codec 子目录中，在此之前，还需要在根目录下使用 go mod init geerpc 初始化项目，方便后续子 package 之间的引用。
进一步，抽象出对消息体进行编解码的接口 Codec，抽象出接口是为了实现不同的 Codec 实例：

1. type Codec interface {
2.         io.Closer
3.         ReadHeader(*Header) error
4.         ReadBody(interface{}) error
5.         Write(*Header, interface{}) error
6. }

复制代码

紧接着，抽象出 Codec 的构造函数，客户端和服务端可以通过 Codec 的 Type 得到构造函数，从而创建 Codec 实例。这部门代码和工厂模式类似，与工厂模式不同的是，返回的是构造函数，而非实例。

1. type NewCodecFunc func(io.ReadWriteCloser) Codec
3. type Type string
5. const (
6.         GobType  Type = "application/gob"
7.         JsonType Type = "application/json" // not implemented
8. )
10. var NewCodecFuncMap map[Type]NewCodecFunc
12. func init() {
13.         NewCodecFuncMap = make(map[Type]NewCodecFunc)
14.         NewCodecFuncMap[GobType] = NewGobCodec
15. }

复制代码

我们定义了 2 种 Codec，Gob 和 Json，但是现实代码中只实现了 Gob 一种，究竟上，2 者的实现非常靠近，甚至只需要把 gob 换成 json 即可。
首先定义 GobCodec 布局体，这个布局体由四部门构成，conn 是由构建函数传入，通常是通过 TCP 或者 Unix 建立 socket 时得到的链接实例，dec 和 enc 对应 gob 的 Decoder 和 Encoder，buf 是为了防止壅闭而创建的带缓冲的 Writer，一样平常这么做能提拔性能。
day1-codec/codec/gob.go

1. package codec
3. import (
4.         "bufio"
5.         "encoding/gob"
6.         "io"
7.         "log"
8. )
10. type GobCodec struct {
11.         conn io.ReadWriteCloser
12.         buf  *bufio.Writer
13.         dec  *gob.Decoder
14.         enc  *gob.Encoder
15. }
17. var _ Codec = (*GobCodec)(nil)
19. func NewGobCodec(conn io.ReadWriteCloser) Codec {
20.         buf := bufio.NewWriter(conn)
21.         return &GobCodec{
22.                 conn: conn,
23.                 buf:  buf,
24.                 dec:  gob.NewDecoder(conn),
25.                 enc:  gob.NewEncoder(buf),
26.         }
27. }

复制代码

  在 Go 语言中，json.NewDecoder 和 json.Unmarshal 都用于将 JSON 数据解析为 Go中的数据布局，但它们有一些区别：
  

• json.NewDecoder 是通过创建一个 Decoder 对象，从一个 io.Reader（如os.Stdin、文件、网络连接等）中读取 JSON 数据并进行解码。
  
• json.Unmarshal 则是直接将 JSON 数据（以字节切片 []byte 或者字符串的情势）解析并映射到指定的数据布局。
  

  使用场景上，如果数据是从一个输入流中读取，通常使用 json.NewDecoder；如果已经有了 JSON 数据的字节切片或字符串，使用json.Unmarshal 会更方便。json.NewEncoder 和 json.Marshal 同理。
  接着实现 ReadHeader、ReadBody、Write 和 Close 方法。

1. func (c *GobCodec) ReadHeader(h *Header) error {
2.         return c.dec.Decode(h)
3. }
5. func (c *GobCodec) ReadBody(body interface{}) error {
6.         return c.dec.Decode(body)
7. }
9. func (c *GobCodec) Write(h *Header, body interface{}) (err error) {
10.         defer func() {
11.                 _ = c.buf.Flush()
12.                 if err != nil {
13.                         _ = c.Close()
14.                 }
15.         }()
16.         if err := c.enc.Encode(h); err != nil {
17.                 log.Println("rpc codec: gob error encoding header:", err)
18.                 return err
19.         }
20.         if err := c.enc.Encode(body); err != nil {
21.                 log.Println("rpc codec: gob error encoding body:", err)
22.                 return err
23.         }
24.         return nil
25. }
27. func (c *GobCodec) Close() error {
28.         return c.conn.Close()
29. }

复制代码

通信过程

客户端与服务端的通信需要协商一些内容，比方 HTTP 报文，分为header和 body 2 部门，body 的格式和长度通过 header 中的 Content-Type 和 Content-Length 指定，服务端通过解析 header 就能够知道如何从 body 中读取需要的信息。对于 RPC 协议来说，这部门协商是需要自主设计的。为了提拔性能，一样平常在报文的最开始会规划固定的字节，来协商相关的信息。比如第1个字节用来体现序列化方式，第2个字节体现压缩方式，第3-6字节体现 header 的长度，7-10 字节体现 body 的长度。
对于 GeeRPC 来说，现在需要协商的唯逐一项内容是消息的编解码方式。我们将这部门信息，放到布局体 Option 中承载。现在，已经进入到服务端的实现阶段了。
day1-codec/server.go

1. package geerpc
3. const MagicNumber = 0x3bef5c
5. type Option struct {
6.         MagicNumber int        // MagicNumber marks this's a geerpc request
7.         CodecType   codec.Type // client may choose different Codec to encode body
8. }
10. var DefaultOption = &Option{
11.         MagicNumber: MagicNumber,
12.         CodecType:   codec.GobType,
13. }

复制代码

一样平常来说，涉及协议协商的这部门信息，需要设计固定的字节来传输的。但是为了实现上更简单，GeeRPC 客户端固定接纳 JSON 编码 Option，后续的 header 和 body 的编码方式由 Option 中的 CodeType 指定，服务端首先使用 JSON 解码 Option，然后通过 Option 的 CodeType 解码剩余的内容。即报文将以这样的情势发送：

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

复制代码

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

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

复制代码

服务端的实现

通信过程已经定义清晰了，那么服务端的实现就比力直接了。
day1-codec/server.go

1. // Server represents an RPC Server.
2. type Server struct{}
4. // NewServer returns a new Server.
5. func NewServer() *Server {
6.         return &Server{}
7. }
9. // DefaultServer is the default instance of *Server.
10. var DefaultServer = NewServer()
12. // Accept accepts connections on the listener and serves requests
13. // for each incoming connection.
14. func (server *Server) Accept(lis net.Listener) {
15.         for {
16.                 conn, err := lis.Accept()
17.                 if err != nil {
18.                         log.Println("rpc server: accept error:", err)
19.                         return
20.                 }
21.                 go server.ServeConn(conn)
22.         }
23. }
25. // Accept accepts connections on the listener and serves requests
26. // for each incoming connection.
27. func Accept(lis net.Listener) { DefaultServer.Accept(lis) }

复制代码

• 首先定义告终构体 Server，没有任何的成员字段。
  
• 实现了 Accept 方式，net.Listener 作为参数，for 循环等候 socket 连接建立，并开启子协程处理，处理过程交给了 ServerConn 方法。
  
• DefaultServer 是一个默认的 Server 实例，主要为了用户使用方便。
  

如果想启动服务，过程黑白常简单的，传入 listener 即可，tcp 协议和 unix 协议都支持。

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

复制代码

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

1. // ServeConn runs the server on a single connection.
2. // ServeConn blocks, serving the connection until the client hangs up.
3. func (server *Server) ServeConn(conn io.ReadWriteCloser) {
4.         defer func() { _ = conn.Close() }()
5.         var opt Option
6.         if err := json.NewDecoder(conn).Decode(&opt); err != nil {
7.                 log.Println("rpc server: options error: ", err)
8.                 return
9.         }
10.         if opt.MagicNumber != MagicNumber {
11.                 log.Printf("rpc server: invalid magic number %x", opt.MagicNumber)
12.                 return
13.         }
14.         f := codec.NewCodecFuncMap[opt.CodecType]
15.         if f == nil {
16.                 log.Printf("rpc server: invalid codec type %s", opt.CodecType)
17.                 return
18.         }
19.         server.serveCodec(f(conn))
20. }
22. // invalidRequest is a placeholder for response argv when error occurs
23. var invalidRequest = struct{}{}
25. func (server *Server) serveCodec(cc codec.Codec) {
26.         sending := new(sync.Mutex) // make sure to send a complete response
27.         wg := new(sync.WaitGroup)  // wait until all request are handled
28.         for {
29.                 req, err := server.readRequest(cc)
30.                 if err != nil {
31.                         if req == nil {
32.                                 break // it's not possible to recover, so close the connection
33.                         }
34.                         req.h.Error = err.Error()
35.                         server.sendResponse(cc, req.h, invalidRequest, sending)
36.                         continue
37.                 }
38.                 wg.Add(1)
39.                 go server.handleRequest(cc, req, sending, wg)
40.         }
41.         wg.Wait()
42.         _ = cc.Close()
43. }

复制代码

serveCodec 的过程非常简单。主要包罗三个阶段

• 读取哀求 readRequest
  
• 处理哀求 handleRequest
  
• 复兴哀求 sendResponse
  

之条件到过，在一次连接中，允许吸收多个哀求，即多个 request header 和 request body，因此这里使用了for无限制地等候哀求的到来，直到发生错误（比方连接被关闭，吸收到的报文有题目等），这里需要注意的点有三个：

• handleRequest 使用了协程并发实行哀求。
  
• 处理哀求是并发的，但是复兴哀求的报文必须是逐个发送的，并发容易导致多个复兴报文交织在一起，客户端无法解析。在这里使用锁(sending)保证。
  
• 尽力而为，只有在 header 解析失败时，才终止循环。
  

1. // request stores all information of a call
2. type request struct {
3.         h            *codec.Header // header of request
4.         argv, replyv reflect.Value // argv and replyv of request
5. }
7. func (server *Server) readRequestHeader(cc codec.Codec) (*codec.Header, error) {
8.         var h codec.Header
9.         if err := cc.ReadHeader(&h); err != nil {
10.                 if err != io.EOF && err != io.ErrUnexpectedEOF {
11.                         log.Println("rpc server: read header error:", err)
12.                 }
13.                 return nil, err
14.         }
15.         return &h, nil
16. }
18. func (server *Server) readRequest(cc codec.Codec) (*request, error) {
19.         h, err := server.readRequestHeader(cc)
20.         if err != nil {
21.                 return nil, err
22.         }
23.         req := &request{h: h}
24.         // TODO: now we don't know the type of request argv
25.         // day 1, just suppose it's string
26.         req.argv = reflect.New(reflect.TypeOf(""))
27.         if err = cc.ReadBody(req.argv.Interface()); err != nil {
28.                 log.Println("rpc server: read argv err:", err)
29.         }
30.         return req, nil
31. }
33. func (server *Server) sendResponse(cc codec.Codec, h *codec.Header, body interface{}, sending *sync.Mutex) {
34.         sending.Lock()
35.         defer sending.Unlock()
36.         if err := cc.Write(h, body); err != nil {
37.                 log.Println("rpc server: write response error:", err)
38.         }
39. }
41. func (server *Server) handleRequest(cc codec.Codec, req *request, sending *sync.Mutex, wg *sync.WaitGroup) {
42.         // TODO, should call registered rpc methods to get the right replyv
43.         // day 1, just print argv and send a hello message
44.         defer wg.Done()
45.         log.Println(req.h, req.argv.Elem())
46.         req.replyv = reflect.ValueOf(fmt.Sprintf("geerpc resp %d", req.h.Seq))
47.         server.sendResponse(cc, req.h, req.replyv.Interface(), sending)
48. }

复制代码

现在还不能判定 body 的类型，因此在readRequest和 handleRequest 中，day1 将 body 作为字符串处理。吸收到哀求，打印 header，并复兴 geerpc resp ${req.h.Seq}。这一部门后续再实现。
main 函数(一个浅易的客户端)

day1 的内容就到此为止了，在这里我们已经实现了一个消息的编解码器 GobCodec，并且客户端与服务端实现了简单的协议交换(protocol exchange)，即允许客户端使用不同的编码方式。同时实现了服务端的雏形，建立连接，读取、处理并复兴客户端的哀求。
接下来，我们就在 main 函数中看看如何使用刚实现的 GeeRPC 吧。
day1-codec/main/main.go

1. package main
3. import (
4.         "encoding/json"
5.         "fmt"
6.         "geerpc"
7.         "geerpc/codec"
8.         "log"
9.         "net"
10.         "time"
11. )
13. func startServer(addr chan string) {
14.         // pick a free port
15.         l, err := net.Listen("tcp", ":0")
16.         if err != nil {
17.                 log.Fatal("network error:", err)
18.         }
19.         log.Println("start rpc server on", l.Addr())
20.         addr <- l.Addr().String()
21.         geerpc.Accept(l)
22. }
24. func main() {
25.         addr := make(chan string)
26.         go startServer(addr)
28.         // in fact, following code is like a simple geerpc client
29.         conn, _ := net.Dial("tcp", <-addr)
30.         defer func() { _ = conn.Close() }()
32.         time.Sleep(time.Second)
33.         // send options
34.         _ = json.NewEncoder(conn).Encode(geerpc.DefaultOption)
35.         cc := codec.NewGobCodec(conn)
36.         // send request & receive response
37.         for i := 0; i < 5; i++ {
38.                 h := &codec.Header{
39.                         ServiceMethod: "Foo.Sum",
40.                         Seq:           uint64(i),
41.                 }
42.                 _ = cc.Write(h, fmt.Sprintf("geerpc req %d", h.Seq))
43.                 _ = cc.ReadHeader(h)
44.                 var reply string
45.                 _ = cc.ReadBody(&reply)
46.                 log.Println("reply:", reply)
47.         }
48. }

复制代码

• 在 startServer 中使用了信道 addr，确保服务端端口监听成功，客户端再发起哀求。
  
• 客户端首先发送 Option 进行协议交换，接下来发送消息头 h := &codec.Header{}，和消息体 geerpc req ${h.Seq}。
  
• 最后解析服务端的响应 reply，并打印出来。
  

实行结果如下：

1. start rpc server on [::]:63662
2. &{Foo.Sum 0 } geerpc req 0
3. reply: geerpc resp 0
4. &{Foo.Sum 1 } geerpc req 1
5. reply: geerpc resp 1
6. &{Foo.Sum 2 } geerpc req 2
7. reply: geerpc resp 2
8. &{Foo.Sum 3 } geerpc req 3
9. reply: geerpc resp 3
10. &{Foo.Sum 4 } geerpc req 4
11. reply: geerpc resp 4

复制代码

原文链接：https://geektutu.com/post/geerpc-day1.html

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。