Go RPC Inside (client)

Go语言标准库能够自带一个rpc框架还是非常给力的，这可以很大程度的降低写后端网络通信服务的门槛，特别是在大规模的分布式系统中，rpc基本是跨机器通信的标配。rpc能够最大程度屏蔽网络细节，让开发者专注在服务功能的开发上面。下面介绍Go语言rpc框架的客户端内部实现.

Go rpc客户端的逻辑很简单，大体上，就是将一个个的调用请求序列化后原子的发送给服务器，然后有一个专门的gorutine等待服务器应答，这个goroutine会将收到的每个应答分发给对应的请求，这就完成了一次rpc调用。

调用入口

func NewClient(conn io.ReadWriteCloser) *Client

func (client *Client) Call(serviceMethod string, args interface{}, reply interface{}) error

使用rpc客户端，首先要NewClient得到一个客户端对象，然后通过这个客户端对象的Call方法去执行远程服务端方法。这里需要注意，NewClient函数不光是创建了一个客户端对象，同时还创建了一个input goroutine. input goroutine是阻塞在连接上，等待服务端的响应。这里Call方法的主要工作是将一个rpc请求发送到服务端，同时放入一个等待队列，等候服务器的响应到来。

发送请求

经过客户端Call方法提交的rpc的请求，最终将调用如下的send方法发送给服务器，这里重点分析一下send的实现。

func (client *Client) send(call *Call) {
	//
	// 入口处就来了一把锁，这把锁显然是为了保证一个请求能够被原子的写入socket中. 毕竟一个
	// rpc连接是会被多个goroutine并发写入的，因此这里需要保证发送请求的原子性。
	// 
	client.sending.Lock()
	defer client.sending.Unlock()

	// 
	// 这里又来了一把锁，这把锁的目标是客户端的pending等待队列，也就是将每个rpc请求放入等待队列
	// 的时候，需要对这个pending队列做并发写保护。
	//
	// Register this call.
	client.mutex.Lock()
	if client.shutdown || client.closing {
		call.Error = ErrShutdown
		client.mutex.Unlock()
		call.done()
		return
	}
	// 
	// pending队列其实是使用map实现的，这里可以看到每个rpc请求都会生存一个唯一递增的seq, 这个
	// seq就是用来标记请求的，这个很像tcp包的seq。
	seq := client.seq
	client.seq++
	client.pending[seq] = call
	client.mutex.Unlock()

	// 
	// 下面就是将一个rpc请求的所有数据 (请求方法名、seq、请求参数等)进行序列化打包，然后发送
	// 出去. 这里主要采用的是Go标准库自带的gob算法进行请求的序列化。
	// 
	// Encode and send the request.
	client.request.Seq = seq
	client.request.ServiceMethod = call.ServiceMethod
	err := client.codec.WriteRequest(&client.request, call.Args)
	if err != nil {
		client.mutex.Lock()
		call = client.pending[seq]
		delete(client.pending, seq)
		client.mutex.Unlock()
		if call != nil {
			call.Error = err
			call.done()
		}
	}
}

通过对发送过程的分析，可以看出一个rpc请求发出去主要有三个过程，第一个过程是将请求放入等待队列，第二个过程就是序列化，最后一个就是写入socket.

send过程最重要的就是两把锁 + seq。复用连接的原子写、并发请求是实现rpc框架的基础。

读取应答

前面提到在创建一个rpc客户端对象的时候，同时会启动一个input goroutine来等待服务器的响应，下面的input方法就是这个goroutine做的所有工作。

func (client *Client) input() {
	var err error
	var response Response
	//
	// 这里的for循环就是永久负责这个连接的响应读取，只有在连接上发生错误后，才会退出。
	// 
	for err == nil {
		//
		// 首先是读响应头, 响应头一般有一个很重要的信息就是正文数据长度，有了这个长度信息，才知道
		// 读多少正文才是一个应答完毕。
		//
		response = Response{}
		err = client.codec.ReadResponseHeader(&response)
		if err != nil {
			break
		}
		// 
		// 这里是一个很重要的步骤，从响应头里取到seq，这个seq就是客户端生成的seq，在上文的send
		// 过程中发送给了服务器，服务器应答的时候，必须将这个seq响应给客户端。只有这样客户端才
		// 知道这个应答是对应pending队列中的那个请求的。
		// 
		// 这里对pending队列枷锁保护，通过seq提取对应的请求call对象。
		//
		seq := response.Seq
		client.mutex.Lock()
		call := client.pending[seq]
		delete(client.pending, seq)
		client.mutex.Unlock()

		//
		// 这个switch里主要就是处理异常以及正常情况读取响应正文了。异常情况，英文解释很详细了。
		// 
		switch {
		case call == nil:
			// We've got no pending call. That usually means that
			// WriteRequest partially failed, and call was already
			// removed; response is a server telling us about an
			// error reading request body. We should still attempt
			// to read error body, but there's no one to give it to.
			err = client.codec.ReadResponseBody(nil)
			if err != nil {
				err = errors.New("reading error body: " + err.Error())
			}
		case response.Error != "":
			// We've got an error response. Give this to the request;
			// any subsequent requests will get the ReadResponseBody
			// error if there is one.
			call.Error = ServerError(response.Error)
			err = client.codec.ReadResponseBody(nil)
			if err != nil {
				err = errors.New("reading error body: " + err.Error())
			}
			call.done()
		default:
			//
			// 开始读取响应的正文，正文放到call中的Reply中去。
			//
			err = client.codec.ReadResponseBody(call.Reply)
			if err != nil {
				call.Error = errors.New("reading body " + err.Error())
			}
			call.done()
		}
	}
	
	//
	// 下面部分的代码都是在处理连接上出错，以及服务端关闭连接等情况的清理工作. 这部分很重要，
	//  否则可能导致一些调用rpc的goroutine永久阻塞等待，不能恢复工作。
	// 
	// Terminate pending calls.
	client.sending.Lock()
	client.mutex.Lock()
	client.shutdown = true
	closing := client.closing
	if err == io.EOF {
		if closing {
			err = ErrShutdown
		} else {
			err = io.ErrUnexpectedEOF
		}
	}
	for _, call := range client.pending {
		call.Error = err
		call.done()
	}
	client.mutex.Unlock()
	client.sending.Unlock()
	if err != io.EOF && !closing {
		log.Println("rpc: client protocol error:", err)
	}
}

不管有多少goroutine在并发写一个rpc连接，等待在连接上读取应答的goroutine只有一个，这个goroutine负责读取连接上的所有响应，然后将响应分发给对应的请求，也就完成了一次rpc请求了。

自定义编码解码器

type ClientCodec interface {
	WriteRequest(*Request, interface{}) error
	ReadResponseHeader(*Response) error
	ReadResponseBody(interface{}) error

	Close() error
}

ClientCodec定义了编码解码器的接口，也就是说你如果要自定义一个编码解码器，只需要实现这个接口就可以了。

WriteRequest做的事情就是将rpc请求的远程方法名、参数等信息进行序列化打包，然后写入socket中。你完全可以按照自己的想法去打包一请求进行发送。 两个Read接口就是从socket读取响应数据包，然后采用相应的算法进行反序列化解包即可。

Go的rpc如果不能自定义编码解码器，那就只能用在客户端和服务器都是Go语言开发的环境中。这就大幅度的降低了其灵活性。能够自定义编码解码器后，理论上就可以用来实现其他rpc的协议，比如：thrift。github的go-thrift项目确实是在Go rpc的基础上通过自定义编码解码器实现的thrift协议。

timeout

Call调用并没有提供超时机制，如果服务器响应太慢，怎么办？死等下去吗？这很不合理。有人说可以自己实现Call方法，不使用rpc包默认提供的Call。代码如下：

func (client *Client) Call(serviceMethod string, args interface{}, reply interface{}) error {
	select {
	case call := <-client.Go(serviceMethod, args, reply, make(chan *Call, 1)).Done:
		return call.Error
	case <-time.After(5*time.Second):
	}
	return errors.New("timeout")
}

上面这段代码确实可以做到rpc调用5s没有得到响应，就超时放弃等待应答。

但是，这样做会有另外一个潜在的风险，就是内存泄漏. 因为，client.Go这个方法会构造一个call对象，则个call对象其实就是一个请求，这个请求被发送出去的同时会放到一个pending队列中，上文已经分析过了。通过分析input goroutine的过程可以看到如果服务器不响应的话，这个call对象就一直待在pending队列中了，永远不会被删除掉。当然，这种情况说明我们的服务器设计有问题。

实现一个rpc客户端，最重要的事情就是连接复用，也就是说一个连接上需要并发的同时跑多个请求。Go rpc在完成这件事情上，动用了两把锁，这个地方很可能会是影响性能的点。待服务器实现分析完后，我们来动手测试一下rpc的性能, 看看究竟如何。