前言
本文是记录在学习 GeeRPC 项目中的一些要点和概念，在代码部分只会给出最核心的代码，而一些外围代码，例如错误处理、一些并发细节、一些接口定义等都会被忽略。适合从一个宏观层面来理解 GeeRPC，因为 GeeRPC 原文实际上是从细节出发，难以从一个宏观视角理解，容易在「森林」中迷路。
完整的源代码还请参考仓库 7daysgolang。
什么是 RPC
RPC 全称 Remote Procedure Call。简单来说，就是将向服务区获取资源这一操作简化成一个函数调用，就像在本地调用函数一样，给入一些参数，返回一个结果。对于使用者来说，他就像本地函数调用一样简单。而本质上这是一次客户端向服务端的网络请求。
一个典型的 RPC 调用如下：
err = client.Call("Service.Method", args, &reply)
其中客户端给定的参数是服务名和方法名的字符串，并给入一组参数 args，服务端会将结果写入 reply，并返回一个错误信息 err。
GeeRPC 的全局视角
从最简化的视角来看，一个 RPC 调用需要一个客户端和服务端，客户端需要发送要调用的方法名，方法的参数。服务端在收到客户端的请求之后，找到要调用的方法，并将参数传入，最终将结果返回给客户端。
从最底层的角度看，客户端需要把调用的方法名、参数序列化成一个二进制序列发送给服务端，而服务端需要解析这个二进制序列，还原成调用的方法名（字符串）、参数（特定的类型）。所以在这部分，至少需要一个编解码器，来对二进制序列进行序列化和还原。实际上这就是一种协议，来规定双方是如何进行交流通信的。
客户端收到返回值（特定类型）之后，还需要再进行一次解码，将数据还原成可读的状态。
在 Go 中，为了简化这种来回编解码的过程，和简化代码接口，将参数和返回值都作为 RPC 调用的参数。其中返回值必须是一个指针，也就是在函数内的修改是可见的，在外部才能获取到这个返回值。而为了符合 Go 的代码设计哲学，返回值代表的是错误信息。
GeeRPC 是大体上参考标准库 net/rpc 的实现，并在此基础上添加了一些额外的功能。
消息的序列化和反序列化
在网络上传输的最终是二进制数据，所以我们要对 Go 中的变量序列化成一个二进制序列，在本项目中使用了 Go 自带的序列化库 gob，其可以将 Go 中的数据类型编码成二进制序列，最终再将这个二进制序列还原成 Go 中的数据类型，一个最简单的示例如下：
str_enc := "Hello World"
var str_dec string
// 创建编解码器
var buffer bytes.Buffer
encoder := gob.NewEncoder(&buffer)
decoder := gob.NewDecoder(&buffer)
// 编码，编码之后二进制序列写入到 buffer 中
encoder.Encode(str_enc)
// 解码，从 buffer 中读取二进制序列解码
decoder.Decode(&str_dec)
在真正编解码之前，我们需要配置一个协议来表示这是一个 RPC 请求，不能只要一个二进制序列到来就进行解析，我们定义一个 Option，其中包含两个字段 MagicNumber 用来表示开始序列，只有以这个序列开始，才表示这是一个 RPC 调用，第二个字段 CodecType 来表示编解码的方法。（也许在未来会添加新的编解码方法，不使用标准库的 gob，提供了一种拓展性）
type Option struct {
MagicNumber int
CodecType string
}
随后，我们定义一个 RPC 调用是怎么构成的，我们可以将 RPC 调用拆分为 header 和 body。其中 header 代表了一些 RPC 调用的元信息，方便客户端和服务端从一开始就能获取到一些关键的信息，例如要调用的方法名、调用序号、错误信息等等。body 则代表 RPC 调用的实际数据，例如参数、返回值等等。
type Header struct {
ServiceMethod string
Seq uint64
Error string
}
不定义 body 是因为 body 就是剩下的二进制序列，没有必要再定义一个。
我们再定义一下一个编解码器所需要的接口：
- 他首先需要一个
Close()函数，因为编解码器的 buffer 实际上来自于网络连接，在发生错误时，他要关闭这个网络连接。 - 然后是
ReadHeader(*Header)和ReadBody(interface{})接口，分别读取 header 和 body。 - 最后实现一个
Write(*Header, interface{})来将二进制数据写入到网络连接中。
接着再定义编解码器的字段：
- 一个已经建立的网络连接
- 一个来自网络连接的缓冲区
- 解码器
- 编码器
所以我们可以编写以下代码：
type GobCodec struct {
conn io.ReadWriteCloser
buf *bufio.Writer
dec *gob.Decoder
enc *gob.Encoder
}
// 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),
}
}
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) Close() error {
return c.conn.Close()
}
func (c *GobCodec) Write(h *Header, body interface{}) (err error) {
defer {
// 调用 Flush() 确保网络数据的发送
c.buf.Flush()
}
// 编码写入缓冲区，实际上会进行网络数据的发送
c.enc.Encode(h)
c.enc.Encode(body)
return
}
需要注意的是，要编解码的类，类和字段只有是导出才能够被编解码。
服务
在完成消息的序列化和反序列之后，我们需要定义，什么是一个服务（service），服务端如何根据到来的信息选择所要调用的方法。当然，最简单的方法是我们可以使用硬解码，例如：
// 服务端收到信息...
if ServiceMethod == "Foo" {
// 调用 Foo
} else if ServiceMethod == "Bar" {
// 调用 Bar
}
// 将结果返回给客户端...
然而这样的方法是不灵活的，如果存在 100 个调用，这个 if-else 就要有 100 个分支，显然不够优雅。在这里我们就要使用到 Go 的反射机制了，反射机制简单来说就是一种动态的解析，假设代码中存在这个方法或者存在这个类，那么 Go 就能找到，不用手动硬编码，例如：
// method 是一个 string 到方法调用的映射
method = make(map[string]reflect.Method)
// 根据反射注册函数，假设 service 是某一个服务类
// 通过反射可以自动获取该类的所有函数
for i := 0; i < service.NumMethod(); i++ {
m = service.Method(i)
method[m.Name] = m
}
// 根据反射找到该函数
m = method[methodName]
if m != nil {
// 如果找到，则调用
m.call(...)
} else {
return errors.New("can't find method " + methodName)
}
对于要调用的函数可以使用反射，对于函数的参数，同样可以使用反射，而不用硬编码，所以定义一个 methodType 来代表一个可调用的函数：
type methodType struct {
method reflect.Method // 函数对象
ArgType reflect.Type // 参数类型
ReplyType reflect.Type // 返回值类型
}
并定义一个 service 来代表一个服务类：由于服务类可能是有状态的（有内部字段），所以需要一个类的实例。
type service struct {
name string // 服务名
typ reflect.Type // 注册服务的类的类型
rcvr reflect.Value // 注册服务的类的实例
method map[string]*methodType // 类的所有方法
}
接着，我们就可以实现其所需要的一些函数，对于 methodType 来说，我们需要一个根据所记录的类型创建变量的函数，methodType 只记录了函数的签名，而没有实例，当我们真正需要构建一个调用时，就需要创建这些参数的实例。
// 创建一个参数类型的实例
func (m *methodType) newArgv() reflect.Value {
var argv reflect.Value
// 指针和值类型需要区分，有细微区别
if m.ArgType.Kind() == reflect.Ptr {
argv = reflect.New(m.ArgType.Elem())
} else {
argv = reflect.New(m.ArgType).Elem()
}
return argv
}
// 创建一个返回值类型的实例
func (m *methodType) newReplyv() reflect.Value {
// 返回值必须是指针
replyv := reflect.New(m.ReplyType.Elem())
// 这两种特殊的数据结构需要初始化
switch m.ReplyType.Elem().Kind() {
case reflect.Map:
replyv.Elem().Set(reflect.MakeMap(m.ReplyType.Elem()))
case reflect.Slice:
replyv.Elem().Set(reflect.MakeSlice(m.ReplyType.Elem(), 0, 0))
}
return replyv
}
对于 service 来说，他需要基于一个给定的服务类实例注册服务，另一个职责则是调用指定的方法：
// 必须保证函数签名是第一个参数是 argType，第二个参数是 replyType
// 这里忽略了一些检查的代码
func (s *service) registerMethods() {
s.method = make(map[string]*methodType)
// 获取所有类的方法
for i := 0; i < s.typ.NumMethod(); i++ {
method := s.typ.Method(i)
mType := method.Type
argType, replyType := mType.In(1), mType.In(2)
s.method[method.Name] = &methodType{
method: method,
ArgType: argType,
ReplyType: replyType,
}
}
}
// 调用方法，需要给定调用的函数签名和参数
func (s *service) call(m *methodType, argv, replyv reflect.Value) error {
// 获取可调用的函数对象
f := m.method.Func
returnValues := f.Call([]reflect.Value{s.rcvr, argv, replyv})
return nil
}
这样，服务端在收到一个 RPC 调用请求之后，解析出函数名字符串，然后在 service 中的 map 中查找，找到其对应的函数前面，并创建对应的变量，将二进制数据解析写入到该变量中，随后调用 call 来进行实际的调用。
服务端
到现在，就可以进行服务端代码的编写了，服务端和客户端是两个职责非常多的类，一方面要使用编解码器进行解析，另一方面还要对可调用的方法进行维护，并考虑各种错误情况。
但是，其最核心的流程可以简化为以下几步：
- 监听可能到来的客户端连接（TCP 连接）
- 建立连接之后，解析并响应来自服务端的请求信息（编解码器和服务反射）
第一步最简单，Go 已经给了非常方便的 tcp 接口来实现监听，对于每个来到的 tcp 连接，我们都建立一个单独的 goroutine 来处理客户端的请求，本文就不再赘述。
第二步解析请求信息的核心步骤可以分为：
- 查看协议头，是否是一个 RPC 请求，如果是，进入一个死循环等待客户端的请求，如果不是，断开连接
- 读取 RPC 请求，分为读取 header 和 body
- 根据 header 在注册的服务中查找，如果没找到，则返回服务未找到错误
- 根据 body 创建相应的参数实例，并调用
service.call来进行实际调用
- 获取结果之后，发送响应给客户端
这里查看协议头的编解码器使用的是 json，和 gob 并没有什么不同，只是编码是将数据结构编码成 json 字符串，解码是根据 json 字符串复原。
在具体进入流程之前，还有一些前期工作要做，第一个是类的定义，第二个是服务的注册：
// 服务端只有一个字段，代表服务名到服务实例的注册
type Server struct {
serviceMap Map[string]*service
}
func (server *Server) Register(rcvr interface{}) error {
s := newService(rcvr)
server.serviceMap.LoadOrStore(s.name, s)
return nil
}
// 创建一个服务
func newService(rcvr interface{}) *service {
s := new(service)
// 根据 rcvr 创建一个相同类型的值
s.rcvr = reflect.ValueOf(rcvr)
// Indirect 会不断解引用，找到最终的那个值
// 假设 rcvr 是多层嵌套的指针，也会找到最终所指向的值
s.name = reflect.Indirect(s.rcvr).Type().Name()
// 获取 rcvr 的类型
s.typ = reflect.TypeOf(rcvr)
s.registerMethods()
return s
}
现在我们就可以到具体的服务器流程中了
func (server *Server) ServeConn(conn io.ReadWriteCloser) {
defer
conn.Close()
var opt Option
// 查看协议头
json.NewDecoder(conn).Decode(&opt)
// 协议头是否满足
if opt.MagicNumber != MagicNumber {
return
}
// 创建编解码器
cc := codec.NewGobCodec(conn)
server.serveCodec(cc)
}
func (server *Server) serveCodec(cc codec.Codec) {
for {
req, _ := server.readRequest(cc)
// 省略请求错误的处理...
go server.handleRequest(cc, req)
}
cc.Close()
}
在完成连接的建立和初步的协议检查之后，就进入了一个死循环，分为读取请求和处理请求，我们先看读取请求：
func (server *Server) readRequest(cc codec.Codec) (*request, error) {
h, _ := server.readRequestHeader(cc)
// 创建 request 实例，并给各个字段赋值
req := &request{h: h}
// 在注册的服务中查找，获取服务类实例和函数签名
req.svc, req.mtype, _ = server.findService(h.ServiceMethod)
req.argv = req.mtype.newArgv()
req.replyv = req.mtype.newReplyv()
// Interface() 的作用是将反射值拷贝并转化类型为 interface{}
argvi := req.argv.Interface()
// 转化为指针之后才能将数据写入到 argvi 中
if req.argv.Type().Kind() != reflect.Ptr {
argvi = req.argv.Addr().Interface()
}
// 写入数据
cc.ReadBody(argvi)
return req, nil
}
func (server *Server) readRequestHeader(cc codec.Codec) (*codec.Header, error) {
var h codec.Header
cc.ReadHeader(&h)
return &h, nil
}
func (server *Server) findService(serviceMethod string) (svc *service, mtype *methodType, err error) {
dot := strings.LastIndex(serviceMethod, ".")
serviceName, methodName := serviceMethod[:dot], serviceMethod[dot+1:]
svci, _ := server.serviceMap.Load(serviceName)
svc = svci.(*service)
mtype = svc.method[methodName]
return svc, mtype, err
}
在读取完请求之后，所有的参数都会在本地创建对应的实例，然后再进行处理请求：
func (server *Server) handleRequest(cc codec.Codec, req *request) {
req.svc.call(req.mtype, req.argv, req.replyv)
server.sendResponse(cc, req.h, req.replyv.Interface(), sending)
}
func (server *Server) sendResponse(cc codec.Codec, h *codec.Header, body interface{}) {
cc.Write(h, body)
}
服务端的主要功能就是这一些，在所给的代码中，忽略了错误检查等等外围代码。
客户端
对于客户端来说，其职责就简单地多，只需要将参数打包，编码成所需的格式，然后再发送到网络即可，大体可以分为以下几步：
- 设置协议头并建立连接
- 将参数打包，指定服务和方法名，编码之后发送请求
- 等待响应，在接收响应之后，解码到来的数据，并在本地创建相应的实例
为了支持并发，geerpc 中使用了一个协程不断等待请求的到来，另一个协程来发送请求，但是简单理解还是可以看作是串行的。
我们首先来看客户端类如何设计，需要什么字段：
// 一个 Call 代表一次调用
type Call struct {
ServiceMethod string // format "<service>.<method>"
Args interface{} // arguments to the function
Reply interface{} // reply from the function
Error error // if error occurs, it will be set
}
type Client struct {
cc codec.Codec
opt *Option
header codec.Header
closing bool // user has called Close
shutdown bool // server has told us to stop
}
这里简化了一些字段，是为了并发实现的，这里忽略。首先建立连接和确认协议。
func NewClient(conn net.Conn, opt *Option) (*Client, error) {
cc := codec.NewGobCodec(conn)
// 将协议头发送给服务器
// 如果协议不对，连接会被服务器中断
json.NewEncoder(conn).Encode(opt)
client := &Client{
cc: cc,
opt: opt,
}
return client, nil
}
随后就要发送请求，这里也删去了并发的一些内容，所以可能看起来函数有点奇怪：
func (client *Client) send(call *Call) {
// 创建请求头
client.header.ServiceMethod = call.ServiceMethod
client.header.Seq = seq
client.header.Error = ""
// 编码并发送数据
client.cc.Write(&client.header, call.Args)
}
最后就是接收响应了，实际上是有另一个死循环不断执行 receive 函数，本文只需要理解核心代码即可。
// 将数据写入到 Call 里
func (client *Client) receive(call *Call) {
var h codec.Header
client.cc.ReadHeader(&h)
client.cc.ReadBody(call.Reply)
}
// 主要流程
type Args struct{ Num1, Num2 int }
fun main() {
conn, _ := net.Dial("tcp", "127.0.0.1:1111")
var opt = &Option{
MagicNumber: MagicNumber,
CodecType: codec.GobType,
}
args := &Args{Num1: i, Num2: i * i}
call := &Call{
ServiceMethod: "Service.Hello",
Args: args,
Reply: &reply,
}
client = NewClient(conn, &opt)
client.send(&call)
client.receive(&call)
reply := call.Reply.(int)
}
到此，geerpc 的主要框架就梳理好了！
外围服务
除了主要框架的基本功能之外，geerpc 还实现了一些其他功能，例如支持 HTTP，负载均衡
HTTP
支持 HTTP 的出发点在于能够让 RPC 更加灵活、可扩展，并易于与现有的 Web 基础设施集成。其原理就是服务端和客户端的第一次信息交换是通过 HTTP 协议进行的，而一旦双方确认之后，后续的信息传输是在一个底层的 TCP 中实现的。
在客户端的修改比较简单，只需要单独增加一个 HTTP 协议的客户端：
func NewHTTPClient(conn net.Conn, opt *Option) (*Client, error) {
_, _ = io.WriteString(conn, fmt.Sprintf("CONNECT %s HTTP/1.0\n\n", defaultRPCPath))
// 确认 HTTP 状态
// resp, err := http.ReadResponse(bufio.NewReader(conn), &http.Request{Method: "CONNECT"})
// ...
return NewClient(conn, opt)
}
此时我们手动构造了一个字符串发送给服务端，服务端会视为是一个 HTTP 请求，并响应一个 HTTP 协议的回复，后续就和原来一样了。
这个关键在于，第一次的 HTTP 交流是为了建立 TCP 连接，在建立了 TCP 连接之后，HTTP 也就不需要了。
// 外围函数：在 Go 中注册 HTTP 响应函数
func (server *Server) HandleHTTP() {
http.Handle(defaultRPCPath, server)
}
// 核心函数，在 HTTP 请求到来之后会调用
func (server *Server) ServeHTTP(w http.ResponseWriter, req *http.Request) {
if req.Method != "CONNECT" {
w.Header().Set("Content-Type", "text/plain; charset=utf-8")
w.WriteHeader(http.StatusMethodNotAllowed)
_, _ = io.WriteString(w, "405 must CONNECT\n")
return
}
// 获取底层 TCP 连接
// 执行 Hijack() 之后就不能再用 http.ResponseWriter 响应请求了
conn, _, _ := w.(http.Hijacker).Hijack()
_, _ = io.WriteString(conn, "HTTP/1.0 "+connected+"\n\n")
// 后续就和原来一模一样了
server.ServeConn(conn)
}
负载均衡
负载均衡的本质就是有多个一样的服务器可以请求，多个客户端尽可能平均地将请求分布在这多个服务器上。geerpc 的负载均衡主要在客户端上实现，服务端并没有做什么修改。
但负载均衡的实现，一方面是可以通过客户端解决，例如在客户端上储存多个服务器的地址，采用一些算法均匀地将请求分布在这些服务器上面。另一方面是可以在服务端设定一些机制，让客户端去请求另一台服务器。
还有一种更方便的方法，建立一个中心节点（代理），负责分发这些请求，所有的客户端请求都只会给这个代理服务器，这个代理服务器来决定最终这个请求要转发给哪一台服务器，这样的好处是易于管理。
在 geerpc 中就只是在客户端做了修改，如果客户端是恶意的，则负载均衡就没有意义了。
func (xc *XClient) Call(ctx context.Context, serviceMethod string, args, reply interface{}) error {
// 根据负载均衡模式，获取下一个服务器的地址
// 最终的调用就会发送给这个地址的服务器
rpcAddr, _ := xc.d.Get(xc.mode)
return xc.call(rpcAddr, ctx, serviceMethod, args, reply)
}
// 最简单的有随机选择和轮转选择
func (d *MultiServersDiscovery) Get(mode SelectMode) (string, error) {
n := len(d.servers)
switch mode {
case RandomSelect:
return d.servers[d.r.Intn(n)], nil
case RoundRobinSelect:
s := d.servers[d.index%n]
d.index = (d.index + 1) % n
return s, nil
}
}
目前来说，客户端需要主动维护所有的服务器列表，但是这样的信息对于客户端来说是难以获得的。更好的方式是客户端只需要知道代理节点，代理节点来维护所有的服务器列表，并且可以移除那些已经下线的服务器。维护服务器列表的核心在于心跳连接，代理节点会定期向所有服务器发送心跳信息，假如什么时候收不到心跳信息，则代表该服务器下线，移除这个服务器地址，不会再发送给客户端。
这部分实际上就是一些额外的业务代码，理解原理之后，就是一些代码的编写，本文就不再赘述了。