【golang 源码分析】第二章 RPC client源码分析

素颜马尾好姑娘i 2022-07-15 15:51 264阅读 0赞

# 第二章 **RPC client源码分析** #

** **

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

client是基于单个socket连接来，靠channel来实现复用连接以及并发的。而临时的调用对象Call都是保存在Client的map中的，对每个call怎么查找，根据seq序列号在请求时候发过去，之后response的时候，client根据返回的seq再反查pending的map结果的。

rpc客户端，最重要的事情就是连接复用，也就是说一个连接上需要并发的同时跑多个请求。

#### 1.1 **结构体** ####

**type **Call **struct **\{   
       ServiceMethod string      // The name of the service and method to call.  
       Args          **interface**\{\} // The argument to the function (\*struct).  
       Reply         **interface**\{\} // The reply from the function (\*struct).  
       Error         error       // After completion, the error status.  
       Done          **chan **\*Call  // Strobes when call is complete.  
\}  
  
**type **Client **struct **\{   
       codec ClientCodec  
  
       reqMutex sync.Mutex // protects following  
       request  Request  
  
       mutex    sync.Mutex // protects following  
       seq      uint64  
       pending  **map**\[uint64\]\*Call  
       closing  bool // user has called Close  
       shutdown bool // server has told us to stop  
\}

**type **ClientCodec **interface **\{   
       // WriteRequest must be safe for concurrent use by multiple goroutines.  
       WriteRequest(\*Request, **interface**\{\}) error  
       ReadResponseHeader(\*Response) error  
       ReadResponseBody(**interface**\{\}) error  
  
       Close() error  
\}

定义了编码解码器的接口，如要自定义一个编码解码器，只要实现这个接口就OK了。WriteRequest就是将rpc请求的远程方法名、参数等信息进行序列化打包，写入[socket][]中。也可以按照自己的意愿去打包请求进行发送。 两个Read接口就是从socket读取响应数据包，然后采用相应的算法进行反序列化解包。

如不能自定义编码[解码器][Link 1]，就只能用在客户端和服务器都是Go。这降低了其灵活性。自定义编码解码器后，就可以实现其他rpc的协议，比如：thrift。github的go-thrift项目。

#### 1.2 **调用入口NewClient** ####

func NewClient(conn io.ReadWriteCloser) \*Client

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

**func **NewClient(conn io.ReadWriteCloser) \*Client \{   
       encBuf := bufio.NewWriter(conn)  
       client := &gobClientCodec\{conn, gob.NewDecoder(conn), gob.NewEncoder(encBuf), encBuf\}  
       **return **NewClientWithCodec(client)  
\}  
**func **NewClientWithCodec(codec ClientCodec) \*Client \{   
       client := &Client\{   
              codec:   codec,  
              pending: make(**map**\[uint64\]\*Call),  
       \}  
       **go **client.input()  
       **return **client  
\}

new一个客户端对象，通过这个对象的Call方法去执行远程[服务端][Link 2]方法。同时建了一个input goroutine. input goroutine阻塞在连接上，wait for服务端的响应。NewClient内部使用的默认的gob编码，gobClientCodes实现了Codec的接口。

Call方法的主要是将一个rpc请求发送到服务端，同时放入一个等待队列，等待[服务器][Link 3]的响应。

#### 1.3 **input函数** ####

**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是客户端生成的，send中发送给[服务器][Link 3]，服务器应答的时候，将这个seq响应给客户端。只有依靠seq客户端才知道这个应答是对应pending队列中的哪个请求。

seq := response.Seq  
              client.mutex.Lock()  
              call := client.pending\[seq\]  
              delete(client.pending, seq)  
              client.mutex.Unlock()  
  
              **switch **\{   
              **case **call == nil:  
                     取出body里的内容并丢弃  
                     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**:  
                     err = client.codec.ReadResponseBody(call.Reply)  
                     **if **err != nil \{   
                            call.Error = errors.New("reading body " \+ err.Error())  
                     \}  
                     call.done()  
              \}  
       \}

下面代码在处理连接上出错，以及[服务端][Link 2]关闭连接等情况的清理工作. 如果不清可能导致一些调用rpc的goroutine永久阻塞等待。

// Terminate pending calls.  
       client.reqMutex.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.reqMutex.Unlock()  
       **if **debugLog && err != io.EOF && !closing \{   
              log.Println("rpc: client protocol error:", err)  
       \}  
\}

等待一个读取应答的goroutine只有一个，这个goroutine负责读取有响应，将响应分发给对应的请求，就完成了一次rpc请求。

#### 1.4 **send函数** ####

**func **(client \*Client) send(call \*Call) \{   
       client.reqMutex.Lock()  
       **defer **client.reqMutex.Unlock()  
  
       // 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()  
  
       // 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][]。

#### 1.5 **接口函数Call** ####

同步死等：发送请求的时候，调用就是Call，传入方法名，参数，获取返回等

**func **(client \*Client) Call(serviceMethod string, args **interface**\{\}, reply **interface**\{\}) error \{   
       call := <-client.Go(serviceMethod, args, reply, make(**chan **\*Call, 1)).Done  
       **return **call.Error  
\}

#### 1.6 **接口函数Go** ####

异步

**func **(client \*Client) Go(serviceMethod string, args **interface**\{\}, reply **interface**\{\}, done **chan **\*Call) \*Call \{   
       call := new(Call)  
       call.ServiceMethod = serviceMethod  
       call.Args = args  
       call.Reply = reply  
       **if **done == nil \{   
              done = make(**chan **\*Call, 10) // buffered.  
       \} **else **\{   
              // If caller passes done != nil, it must arrange that  
              // done has enough buffer for the number of simultaneous  
              // RPCs that will be using that channel. If the channel  
              // is totally unbuffered, it's best not to run at all.  
              **if **cap(done) == 0 \{   
                     log.Panic("rpc: done channel is unbuffered")  
              \}  
       \}  
       call.Done = done  
       client.send(call)  
       **return **call  
\}

[socket]: http://cpro.baidu.com/cpro/ui/uijs.php?adclass=0&app_id=0&c=news&cf=1001&ch=0&di=128&fv=20&is_app=0&jk=417033db4375a4f0&k=socket&k0=socket&kdi0=0&luki=8&mcpm=0&n=10&p=baidu&q=74042097_cpr&rb=0&rs=1&seller_id=1&sid=f0a47543db337041&ssp2=1&stid=9&t=tpclicked3_hc&td=1989498&tu=u1989498&u=http://studygolang.com/articles/1788&urlid=0
[Link 1]: http://cpro.baidu.com/cpro/ui/uijs.php?adclass=0&app_id=0&c=news&cf=1001&ch=0&di=128&fv=20&is_app=0&jk=417033db4375a4f0&k=%BD%E2%C2%EB%C6%F7&k0=%BD%E2%C2%EB%C6%F7&kdi0=0&luki=10&mcpm=0&n=10&p=baidu&q=74042097_cpr&rb=0&rs=1&seller_id=1&sid=f0a47543db337041&ssp2=1&stid=9&t=tpclicked3_hc&td=1989498&tu=u1989498&u=http://studygolang.com/articles/1788&urlid=0
[Link 2]: http://cpro.baidu.com/cpro/ui/uijs.php?adclass=0&app_id=0&c=news&cf=1001&ch=0&di=128&fv=20&is_app=0&jk=417033db4375a4f0&k=%B7%FE%CE%F1%B6%CB&k0=%B7%FE%CE%F1%B6%CB&kdi0=0&luki=5&mcpm=0&n=10&p=baidu&q=74042097_cpr&rb=0&rs=1&seller_id=1&sid=f0a47543db337041&ssp2=1&stid=9&t=tpclicked3_hc&td=1989498&tu=u1989498&u=http://studygolang.com/articles/1788&urlid=0
[Link 3]: http://cpro.baidu.com/cpro/ui/uijs.php?adclass=0&app_id=0&c=news&cf=1001&ch=0&di=128&fv=20&is_app=0&jk=417033db4375a4f0&k=%B7%FE%CE%F1%C6%F7&k0=%B7%FE%CE%F1%C6%F7&kdi0=0&luki=9&mcpm=0&n=10&p=baidu&q=74042097_cpr&rb=0&rs=1&seller_id=1&sid=f0a47543db337041&ssp2=1&stid=9&t=tpclicked3_hc&td=1989498&tu=u1989498&u=http://studygolang.com/articles/1788&urlid=0