Golang手写RPC框架(day1)

我的项目地址 包括项目源码和笔记内容。

Day1 服务端与消息编码

具体流程见

传送门

这里只介绍一下额外的细节和其他知识点。

1.多主程序项目的启动

Goland对于多主程序的项目，可以使用不同的配置来启动不同的项目。

build目录是 使用go build 后程序生成的exe可执行文件。

run是工作目录，可以进行额外的配置。(这里没有作用)

2.流程

这里说明主程序的流程。

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

3.细节

监听端口

l, err := net.Listen("tcp", ":0")

这里:0是随机监听端口。

比如在我的机器上就是13519。

json编码解码

_ = json.NewEncoder(conn).Encode(geerpc.DefaultOption)

这里使用json库 将Option：geerpc.DefaultOption 写入到conn结构体的Writer里，表示conn的相关选项Option设置。

使用方法类似于下面的：

 // 3. 使用 json.NewEncoder 编码
    person3 := Person{"王五", 30}
    // 编码结果暂存到 buffer
    bytes3 := new(bytes.Buffer)
    _ = json.NewEncoder(bytes3).Encode(person3)
    if err == nil {
        fmt.Print("json.NewEncoder 编码结果: ", string(bytes3.Bytes()))
    }


// json.NewEncoder 编码结果: {"name":"王五","age":30}

创建自定义的编码解码器

cc := codec.NewGobCodec(conn)

func NewGobCodec(conn io.ReadWriteCloser) Codec {
	buf := bufio.NewWriter(conn)
	return &GobCodec{
		conn: conn,
		buf:  buf,
		dec:  gob.NewDecoder(conn),
		enc:  gob.NewEncoder(buf),
	}
}

这里使用buffer缓存提高写入的效率，也就是说需要写入的内容先写入到buf中，然后用buf.flush() 写到conn的Writer中，而读取不需要缓存，所以直接使用

gob.NewDecoder(conn) 作为解码器读取。

服务器端处理请求

func (server *Server) serveCodec(cc codec.Codec) {
	sending := new(sync.Mutex) // make sure to send a complete response
	wg := new(sync.WaitGroup)  // wait until all request are handled
	for {
		req, err := server.readRequest(cc) //获取请求的head 和 body
		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) //处理请求并返回req.replyv
	}
	wg.Wait()
	_ = cc.Close()
}

使用sending互斥锁 保证服务器端按顺序响应请求，不会同时响应多个请求。

处理请求是并发的，但是回复请求的报文必须是逐个发送的，并发容易导致多个回复报文交织在一起，客户端无法解析。在这里使用锁(sending)保证。

使用sync.WaitGroup保证处理完所有请求后再关闭conn。

使用Codec进行读写操作

_ = cc.Write(h, fmt.Sprintf("geerpc req %d", h.Seq))

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 {
		log.Println("rpc: gob error encoding body:", err)
		return
	}
	return
}

func (c *GobCodec) ReadHeader(h *Header) error {
	return c.dec.Decode(h)
}

func (c *GobCodec) ReadBody(body interface{}) error {
	return c.dec.Decode(body)
}

这里conn 拥有一块内存区域，用来读写。 c.enc.Encode 将内容编码到buf里缓存，然后flush到conn里。然后读操作ReadHeader和Body时，使用decoder将conn里的内容解码。

验证接口实现

var _ Codec = (*GobCodec)(nil)

用于验证GobCodec是否实现了Codec接口。

另外这个项目实现Codec接口的工厂模式，不同类型实例有不同的构造函数，比较有价值，可以效仿。

4.输出结果