blog
blog copied to clipboard
geektutu
动手写RPC框架 - GeeRPC第一天 服务端与消息编码 | 极客兔兔
https://geektutu.com/post/geerpc-day1.html
7天用 Go语言/golang 从零实现 RPC 框架 GeeRPC 教程(7 days implement golang remote procedure call framework from scratch tutorial)，动手写 RPC 框架，参照 golang 标准库 net/rpc 的实现，实现了服务端(server)、支持异步和并发的客户端(client)、消息编码与解码(message encoding and decoding)、服务注册(service register)、支持 TCP/Unix/HTTP 等多种传输协议。第一天实现了一个简单的服务端和消息的编码与解码。
直接这样可以吗defer func() { conn.Close() }() 而不是 defer func() { _ = conn.Close() }()
client.sending.Lock() defer client.sending.Unlock() client.mu.Lock() defer client.mu.Unlock() 问题1：不太理解为啥terminateCalls方法，需要sending和mu两把锁，而注册call和删除call，只需要mu锁？ 问题2：当client的某一个字段比如sending Lock()时，是不是在unlock()之前，这个client对象的sending字段是线程安全的，不会被其他线程或协程访问到吗？
@leigexiaohuozi 直接这样可以吗defer func() { conn.Close() }() 而不是 defer func() { _ = conn.Close() }()
应该是 defer conn.Close()
刚才照着大佬的代码敲了一遍，发现main函数中的循环发送请求如果不执行readBody或者执行出错，我们的sendResponse就会只打印其中的两三个请求，有时甚至还会报EOF错误：
代码如下：
for 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)
}
是什么原因阻塞了我们的请求呢，又或者执行什么超时导致主线程退出了
@XiaoyeFang 刚才照着大佬的代码敲了一遍，发现main函数中的循环发送请求如果不执行readBody或者执行出错，我们的sendResponse就会只打印其中的两三个请求，有时甚至还会报EOF错误：
代码如下：
for 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) }是什么原因阻塞了我们的请求呢，又或者执行什么超时导致主线程退出了
因为你的主协程 发送完5次请求后就退出了 此时运行服务器的协程还没有打印完这5次请求也退出了
我直接使用的day-1的代码，在windows上可以运行，到我虚拟机的ubuntu上，会阻塞在readRequestHeader，这可能是什么原因呢？
运行到箭头就阻塞了，不知道为啥？
第一个箭头应该是option，第二个是header和body，client的request应该是发出去了？
在windows上没有问题。
@yangchen97 我想问问如果不在header里指定body长度，会有粘包拆包的问题吗？
json 字符串是有数据的边界的即 "{" 和 "}"所以这里并不会出现粘包的问题
问下 这行代码 // send options _ = json.NewEncoder(conn).Encode(geerpc.DefaultOption) 是怎么跟server通讯把option发过去的啊，我看这个conn就是conn, _ := net.Dial("tcp", <-addr).
@wilgx0
@yangchen97 我想问问如果不在header里指定body长度，会有粘包拆包的问题吗？
json 字符串是有数据的边界的即 "{" 和 "}"所以这里并不会出现粘包的问题
学到了
func NewGobCodec(conn io.ReadWriteCloser) Codec {
buf := bufio.NewWriter(conn)
return &GobCodec{
conn: conn,
buf: buf,
dec: gob.NewDecoder(conn),
enc: gob.NewEncoder(buf),
}
}
秒啊enc使用新的buffer，我一开始没注意使用的gob.NewEncoder(conn),导致可能会Option和header一起传过去，就会导致用body的解析为header就报错了
@sam-lc
func NewGobCodec(conn io.ReadWriteCloser) Codec { buf := bufio.NewWriter(conn) return &GobCodec{ conn: conn, buf: buf, dec: gob.NewDecoder(conn), enc: gob.NewEncoder(buf), } }秒啊enc使用新的buffer，我一开始没注意使用的gob.NewEncoder(conn),导致可能会Option和header一起传过去，就会导致用body的解析为header就报错了
请问这里可以解释一下，没太明白buf := bufio.NewWriter(conn)以及enc: gob.NewEncoder(buf),这两句
@4ttenji
@sam-lc
func NewGobCodec(conn io.ReadWriteCloser) Codec { buf := bufio.NewWriter(conn) return &GobCodec{ conn: conn, buf: buf, dec: gob.NewDecoder(conn), enc: gob.NewEncoder(buf), } }秒啊enc使用新的buffer，我一开始没注意使用的gob.NewEncoder(conn),导致可能会Option和header一起传过去，就会导致用body的解析为header就报错了
请问这里可以解释一下，没太明白buf := bufio.NewWriter(conn)以及enc: gob.NewEncoder(buf),这两句
encoder 因为是要往 conn 中写入内容, 这里文章中说明了要使用 buffer 来优化写入效率, 所以我们先写入到 buffer 中, 然后我们再调用 buffer.Flush() 来将 buffer 中的全部内容写入到 conn 中, 从而优化效率. 对于读则不需要这方面的考虑, 所以直接在 conn 中读内容即可.
type GobCodec struct { conn io.ReadWriteCloser buf *bufio.Writer dec *gob.Decoder enc *gob.Encoder } codec.GobCodec on pkg.go.dev
cannot use (*GobCodec)(nil) (value of type *GobCodec) as Codec value in variable declaration: missing method
json包里面的encode 和 decode只是简单调用的net包的read和write，没有处理tcp的数据边界问题，这样不会有问题吗？
即使json有{}的分隔符，但是json.decode里面代码是调用了底层conn.read(),这里可能会把header里面的数据读出来，导致下次读取header数据出现残缺。
@lazzman
@valiner
json包里面的encode 和 decode只是简单调用的net包的read和write，没有处理tcp的数据边界问题，这样不会有问题吗？
即使json有{}的分隔符，但是json.decode里面代码是调用了底层conn.read(),这里可能会把header里面的数据读出来，导致下次读取header数据出现残缺。
个人见解，不对请指正！ 见
/sdk/go1.16.4/src/encoding/json/stream.go:49中Decode方法的实现 从方法注释和代码中的注释能看出来工作机制是从缓冲区中不断读取下一个Json编码内容 每次反序列前会从conn中读取所有的数据到缓冲区中，再从缓冲区数据中读取一个完整的Json编码内容，所以消息粘包问题被golang的这种流式编解码机制解决了。// Decode reads the next JSON-encoded value from its // input and stores it in the value pointed to by v. // // See the documentation for Unmarshal for details about // the conversion of JSON into a Go value. func (dec *Decoder) Decode(v interface{}) error { if dec.err != nil { return dec.err } if err := dec.tokenPrepareForDecode(); err != nil { return err } if !dec.tokenValueAllowed() { return &SyntaxError{msg: "not at beginning of value", Offset: dec.InputOffset()} } // Read whole value into buffer. n, err := dec.readValue() if err != nil { return err } dec.d.init(dec.buf[dec.scanp : dec.scanp+n]) dec.scanp += n // Don't save err from unmarshal into dec.err: // the connection is still usable since we read a complete JSON // object from it before the error happened. err = dec.d.unmarshal(v) // fixup token streaming state dec.tokenValueEnd() return err }
每次反序列前会从conn中读取所有的数据到缓冲区中,这个是时候是不是有可能读到header里面的信息。