本文代碼地址：

本文是7天用Go從零實(shí)現(xiàn)RPC框架GeeRPC的第一篇。

• 使用 encoding/gob 實(shí)現(xiàn)消息的編解碼(序列化與反序列化)
• 實(shí)現(xiàn)一個(gè)簡(jiǎn)易的服務(wù)端，僅接受消息，不處理，代碼約 200 行

消息的序列化與反序列化

一個(gè)典型的 RPC 調(diào)用如下：

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

客戶端發(fā)送的請(qǐng)求包括服務(wù)名 Arith，方法名 Multiply，參數(shù) args 三個(gè)，服務(wù)端的響應(yīng)包括錯(cuò)誤 error，返回值 reply 2 個(gè)。我們將請(qǐng)求和響應(yīng)中的參數(shù)和返回值抽象為 body，剩余的信息放在 header 中，那么就可以抽象出數(shù)據(jù)結(jié)構(gòu) Header：

day1-codec/codec/codec.go

package codecimport "io"type Header struct {ServiceMethod string // format "Service.Method"Seq           uint64 // sequence number chosen by clientError         string
}

• ServiceMethod 是服務(wù)名和方法名，通常與 Go 語(yǔ)言中的結(jié)構(gòu)體和方法相映射。
• Seq 是請(qǐng)求的序號(hào)，也可以認(rèn)為是某個(gè)請(qǐng)求的 ID，用來(lái)區(qū)分不同的請(qǐng)求。
• Error 是錯(cuò)誤信息，客戶端置為空，服務(wù)端如果如果發(fā)生錯(cuò)誤，將錯(cuò)誤信息置于 Error 中。

我們將和消息編解碼相關(guān)的代碼都放到 codec 子目錄中，在此之前，還需要在根目錄下使用 go mod init geerpc 初始化項(xiàng)目，方便后續(xù)子 package 之間的引用。

進(jìn)一步，抽象出對(duì)消息體進(jìn)行編解碼的接口 Codec，抽象出接口是為了實(shí)現(xiàn)不同的 Codec 實(shí)例：

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

緊接著，抽象出 Codec 的構(gòu)造函數(shù)，客戶端和服務(wù)端可以通過(guò) Codec 的 Type 得到構(gòu)造函數(shù)，從而創(chuàng)建 Codec 實(shí)例。這部分代碼和工廠模式類似，與工廠模式不同的是，返回的是構(gòu)造函數(shù)，而非實(shí)例。

type NewCodecFunc func(io.ReadWriteCloser) Codectype Type stringconst (GobType  Type = "application/gob"JsonType Type = "application/json" // not implemented
)var NewCodecFuncMap map[Type]NewCodecFuncfunc init() {NewCodecFuncMap = make(map[Type]NewCodecFunc)NewCodecFuncMap[GobType] = NewGobCodec
}

我們定義了 2 種 Codec，Gob 和 Json，但是實(shí)際代碼中只實(shí)現(xiàn)了 Gob 一種，事實(shí)上，2 者的實(shí)現(xiàn)非常接近，甚至只需要把 gob 換成 json 即可。

首先定義 GobCodec 結(jié)構(gòu)體，這個(gè)結(jié)構(gòu)體由四部分構(gòu)成，conn 是由構(gòu)建函數(shù)傳入，通常是通過(guò) TCP 或者 Unix 建立 socket 時(shí)得到的鏈接實(shí)例，dec 和 enc 對(duì)應(yīng) gob 的 Decoder 和 Encoder，buf 是為了防止阻塞而創(chuàng)建的帶緩沖的 Writer，一般這么做能提升性能。

day1-codec/codec/gob.go

package codecimport ("bufio""encoding/gob""io""log"
)type GobCodec struct {conn io.ReadWriteCloserbuf  *bufio.Writerdec  *gob.Decoderenc  *gob.Encoder
}var _ Codec = (*GobCodec)(nil)func NewGobCodec(conn io.ReadWriteCloser) Codec {buf := bufio.NewWriter(conn)return &GobCodec{conn: conn,buf:  buf,dec:  gob.NewDecoder(conn),enc:  gob.NewEncoder(buf),}
}

在 Go 語(yǔ)言中，json.NewDecoder 和 json.Unmarshal 都用于將 JSON 數(shù)據(jù)解析為 Go中的數(shù)據(jù)結(jié)構(gòu)，但它們有一些區(qū)別：

• json.NewDecoder 是通過(guò)創(chuàng)建一個(gè) Decoder 對(duì)象，從一個(gè) io.Reader（如os.Stdin、文件、網(wǎng)絡(luò)連接等）中讀取 JSON 數(shù)據(jù)并進(jìn)行解碼。
• json.Unmarshal 則是直接將 JSON 數(shù)據(jù)（以字節(jié)切片 []byte 或者字符串的形式）解析并映射到指定的數(shù)據(jù)結(jié)構(gòu)。

使用場(chǎng)景上，如果數(shù)據(jù)是從一個(gè)輸入流中讀取，通常使用 json.NewDecoder；如果已經(jīng)有了 JSON 數(shù)據(jù)的字節(jié)切片或字符串，使用json.Unmarshal 會(huì)更方便。json.NewEncoder 和 json.Marshal 同理。

接著實(shí)現(xiàn) ReadHeader、ReadBody、Write 和 Close 方法。

func (c *GobCodec) ReadHeader(h *Header) error {return c.dec.Decode(h)
}func (c *GobCodec) ReadBody(body interface{}) error {return c.dec.Decode(body)
}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 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
}func (c *GobCodec) Close() error {return c.conn.Close()
}

通信過(guò)程

客戶端與服務(wù)端的通信需要協(xié)商一些內(nèi)容，例如 HTTP 報(bào)文，分為header和 body 2 部分，body 的格式和長(zhǎng)度通過(guò) header 中的 Content-Type 和 Content-Length 指定，服務(wù)端通過(guò)解析 header 就能夠知道如何從 body 中讀取需要的信息。對(duì)于 RPC 協(xié)議來(lái)說(shuō)，這部分協(xié)商是需要自主設(shè)計(jì)的。為了提升性能，一般在報(bào)文的最開(kāi)始會(huì)規(guī)劃固定的字節(jié)，來(lái)協(xié)商相關(guān)的信息。比如第1個(gè)字節(jié)用來(lái)表示序列化方式，第2個(gè)字節(jié)表示壓縮方式，第3-6字節(jié)表示 header 的長(zhǎng)度，7-10 字節(jié)表示 body 的長(zhǎng)度。

對(duì)于 GeeRPC 來(lái)說(shuō)，目前需要協(xié)商的唯一一項(xiàng)內(nèi)容是消息的編解碼方式。我們將這部分信息，放到結(jié)構(gòu)體 Option 中承載。目前，已經(jīng)進(jìn)入到服務(wù)端的實(shí)現(xiàn)階段了。

day1-codec/server.go

package geerpcconst MagicNumber = 0x3bef5ctype Option struct {MagicNumber int        // MagicNumber marks this's a geerpc requestCodecType   codec.Type // client may choose different Codec to encode body
}var DefaultOption = &Option{MagicNumber: MagicNumber,CodecType:   codec.GobType,
}

一般來(lái)說(shuō)，涉及協(xié)議協(xié)商的這部分信息，需要設(shè)計(jì)固定的字節(jié)來(lái)傳輸?shù)?。但是為了?shí)現(xiàn)上更簡(jiǎn)單，GeeRPC 客戶端固定采用 JSON 編碼 Option，后續(xù)的 header 和 body 的編碼方式由 Option 中的 CodeType 指定，服務(wù)端首先使用 JSON 解碼 Option，然后通過(guò) Option 的 CodeType 解碼剩余的內(nèi)容。即報(bào)文將以這樣的形式發(fā)送：

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

在一次連接中，Option 固定在報(bào)文的最開(kāi)始，Header 和 Body 可以有多個(gè)，即報(bào)文可能是這樣的。

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

服務(wù)端的實(shí)現(xiàn)

通信過(guò)程已經(jīng)定義清楚了，那么服務(wù)端的實(shí)現(xiàn)就比較直接了。

day1-codec/server.go

// Server represents an 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 serves 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) }

• 首先定義了結(jié)構(gòu)體 Server，沒(méi)有任何的成員字段。
• 實(shí)現(xiàn)了 Accept 方式，net.Listener 作為參數(shù)，for 循環(huán)等待 socket 連接建立，并開(kāi)啟子協(xié)程處理，處理過(guò)程交給了 ServerConn 方法。
• DefaultServer 是一個(gè)默認(rèn)的 Server 實(shí)例，主要為了用戶使用方便。

如果想啟動(dòng)服務(wù)，過(guò)程是非常簡(jiǎn)單的，傳入 listener 即可，tcp 協(xié)議和 unix 協(xié)議都支持。

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

ServeConn 的實(shí)現(xiàn)就和之前討論的通信過(guò)程緊密相關(guān)了，首先使用 json.NewDecoder 反序列化得到 Option 實(shí)例，檢查 MagicNumber 和 CodeType 的值是否正確。然后根據(jù) CodeType 得到對(duì)應(yīng)的消息編解碼器，接下來(lái)的處理交給 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 request 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 的過(guò)程非常簡(jiǎn)單。主要包含三個(gè)階段

• 讀取請(qǐng)求 readRequest
• 處理請(qǐng)求 handleRequest
• 回復(fù)請(qǐng)求 sendResponse

之前提到過(guò)，在一次連接中，允許接收多個(gè)請(qǐng)求，即多個(gè) request header 和 request body，因此這里使用了for無(wú)限制地等待請(qǐng)求的到來(lái)，直到發(fā)生錯(cuò)誤（例如連接被關(guān)閉，接收到的報(bào)文有問(wèn)題等），這里需要注意的點(diǎn)有三個(gè)：

• handleRequest 使用了協(xié)程并發(fā)執(zhí)行請(qǐng)求。
• 處理請(qǐng)求是并發(fā)的，但是回復(fù)請(qǐng)求的報(bào)文必須是逐個(gè)發(fā)送的，并發(fā)容易導(dǎo)致多個(gè)回復(fù)報(bào)文交織在一起，客戶端無(wú)法解析。在這里使用鎖(sending)保證。
• 盡力而為，只有在 header 解析失敗時(shí)，才終止循環(huán)。

// request stores all information of a call
type request struct {h            *codec.Header // header of requestargv, replyv reflect.Value // argv and replyv of request
}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 error:", 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 中，day1 將 body 作為字符串處理。接收到請(qǐng)求，打印 header，并回復(fù) geerpc resp ${req.h.Seq}。這一部分后續(xù)再實(shí)現(xiàn)。

main 函數(shù)(一個(gè)簡(jiǎn)易的客戶端)

day1 的內(nèi)容就到此為止了，在這里我們已經(jīng)實(shí)現(xiàn)了一個(gè)消息的編解碼器 GobCodec，并且客戶端與服務(wù)端實(shí)現(xiàn)了簡(jiǎn)單的協(xié)議交換(protocol exchange)，即允許客戶端使用不同的編碼方式。同時(shí)實(shí)現(xiàn)了服務(wù)端的雛形，建立連接，讀取、處理并回復(fù)客戶端的請(qǐng)求。

接下來(lái)，我們就在 main 函數(shù)中看看如何使用剛實(shí)現(xiàn)的 GeeRPC 吧。

day1-codec/main/main.go

package mainimport ("encoding/json""fmt""geerpc""geerpc/codec""log""net""time"
)func startServer(addr chan string) {// pick a free portl, 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)// in fact, following code is like a simple geerpc clientconn, _ := 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)}
}

• 在 startServer 中使用了信道 addr，確保服務(wù)端端口監(jiān)聽(tīng)成功，客戶端再發(fā)起請(qǐng)求。
• 客戶端首先發(fā)送 Option 進(jìn)行協(xié)議交換，接下來(lái)發(fā)送消息頭 h := &codec.Header{}，和消息體 geerpc req ${h.Seq}。
• 最后解析服務(wù)端的響應(yīng) reply，并打印出來(lái)。

執(zhí)行結(jié)果如下：

start rpc server on [::]:63662
&{Foo.Sum 0 } geerpc req 0
reply: geerpc resp 0
&{Foo.Sum 1 } geerpc req 1
reply: geerpc resp 1
&{Foo.Sum 2 } geerpc req 2
reply: geerpc resp 2
&{Foo.Sum 3 } geerpc req 3
reply: geerpc resp 3
&{Foo.Sum 4 } geerpc req 4
reply: geerpc resp 4

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