文章目录
- Call 的设计
- 实现 Client
- Demo
本文代码地址：
本文是7天用Go从零实现RPC框架GeeRPC的第二篇。
- 实现一个支持异步和并发的高性能客户端，代码约
250行
Call 的设计
对 net/rpc 而言，一个函数需要能够被远程调用，需要满足如下五个条件：
- the method’s type is exported.
- the method is exported.
- the method has two arguments, both exported (or builtin) types.
- the method’s second argument is a pointer.
- the method has return type error.
更直观一些：
func (t *T) MethodName(argType T1, replyType *T2) error
根据上述要求，首先我们封装了结构体 Call 来承载一次 RPC 调用所需要的信息。
day2-client/client.go
// Call represents an active RPC.
// Call 承载一次RPC调用所需要的信息
type Call struct {Seq uint64ServiceMethod string // format "<service>.<method>"Args interface{} // arguments to the functionReply interface{} // reply from the functionError error // if error occurs, it will be setDone chan *Call // Strobes when call is complete.
}func (call *Call) done() {call.Done <- call
}
为了支持异步调用，Call 结构体中添加了一个字段 Done，Done 的类型是 chan *Call，当调用结束时，会调用 call.done() 通知调用方。
实现 Client
接下来，我们将实现 GeeRPC 客户端最核心的部分 Client。
// Client 表示一个RPC客户端，一个Client可以关联多个Call，即一个客户端可以发起多个RPC调用
// 一个Client也可能同时被多个协程使用
type Client struct {cc codec.Codecopt *Optionsending sync.Mutex // protect followingheader codec.Headermu sync.Mutex // protect followingseq uint64pending map[uint64]*Callclosing bool // user has called Closeshutdown bool // server has told us to stop
}var _ io.Closer = (*Client)(nil)var ErrShutdown = errors.New("connection is shut down")// Close the connection
func (client *Client) Close() error {client.mu.Lock()defer client.mu.Unlock()if client.closing {return ErrShutdown}client.closing = truereturn client.cc.Close()
}// IsAvailable return true if the client does work
func (client *Client) IsAvailable() bool {client.mu.Lock()defer client.mu.Unlock()return !client.shutdown && !client.closing
}
Client 的字段比较复杂：
cc是消息的编解码器，和服务端类似，用来序列化将要发送出去的请求，以及反序列化接收到的响应。sending是一个互斥锁，和服务端类似，为了保证请求的有序发送，即防止出现多个请求报文混淆。header是每个请求的消息头，header只有在请求发送时才需要，而请求发送是互斥的，因此每个客户端只需要一个，声明在Client结构体中可以复用。seq用于给发送的请求编号，每个请求拥有唯一编号。pending存储未处理完的请求，键是编号，值是Call实例。closing和shutdown任意一个值置为true，则表示Client处于不可用的状态，但有些许的差别，closing是用户主动关闭的，即调用Close方法，而shutdown置为true一般是有错误发生。
紧接着，实现和 Call 相关的三个方法。
func (client *Client) registerCall(call *Call) (uint64, error) {client.mu.Lock()defer client.mu.Unlock()if client.closing || client.shutdown {return 0, ErrShutdown}call.Seq = client.seqclient.pending[call.Seq] = callclient.seq++return call.Seq, nil
}func (client *Client) removeCall(seq uint64) *Call {client.mu.Lock()defer client.mu.Unlock()call := client.pending[seq]delete(client.pending, seq)return call
}func (client *Client) terminateCalls(err error) {client.sending.Lock()defer client.sending.Unlock()client.mu.Lock()defer client.mu.Unlock()client.shutdown = truefor _, call := range client.pending {call.Error = errcall.done()}
}
registerCall：将参数call添加到client.pending中，并更新client.seq。removeCall：根据seq，从client.pending中移除对应的call，并返回。terminateCalls：服务端或客户端发生错误时调用，将shutdown设置为true，且将错误信息通知所有pending状态的call。
对一个客户端端来说，接收响应、发送请求是最重要的 2 个功能。那么首先实现接收功能，接收到的响应有三种情况：
call不存在，可能是请求没有发送完整，或者因为其他原因被取消，但是服务端仍旧处理了。call存在，但服务端处理出错，即h.Error不为空。call存在，服务端处理正常，那么需要从body中读取Reply的值。
func (client *Client) receive() {var err errorfor err == nil {var h codec.Headerif err = client.cc.ReadHeader(&h); err != nil {break}call := client.removeCall(h.Seq)switch {case call == nil:// it usually means that Write partially failed// and call was already removed.err = client.cc.ReadBody(nil)case h.Error != "":call.Error = fmt.Errorf(h.Error)err = client.cc.ReadBody(nil)call.done()default:err = client.cc.ReadBody(call.Reply)if err != nil {call.Error = errors.New("reading body " + err.Error())}call.done()}}// error occurs, so terminateCalls pending callsclient.terminateCalls(err)
}
创建 Client 实例时，首先需要完成一开始的协议交换，即发送 Option 信息给服务端。协商好消息的编解码方式之后，再创建一个子协程调用 receive() 接收响应。
func NewClient(conn net.Conn, opt *Option) (*Client, error) {f := codec.NewCodecFuncMap[opt.CodecType]if f == nil {err := fmt.Errorf("invalid codec type %s", opt.CodecType)log.Println("rpc client: codec error:", err)return nil, err}// 将opt发送给服务端if err := json.NewEncoder(conn).Encode(opt); err != nil {log.Println("rpc client: options error: ", err)_ = conn.Close()return nil, err}return newClientCodec(f(conn), opt), nil
}func newClientCodec(cc codec.Codec, opt *Option) *Client {client := &Client{seq: 1, // seq starts with 1, 0 means invalid callcc: cc,opt: opt,pending: make(map[uint64]*Call),}// 启动接收服务端响应的协程go client.receive()return client
}
还需要实现 Dial 函数，便于用户传入服务端地址，创建 Client 实例。为了简化用户调用，通过 ...*Option 将 Option 实现为可选参数。
func parseOptions(opts ...*Option) (*Option, error) {// if opts is nil or pass nil as parameterif len(opts) == 0 || opts[0] == nil {return DefaultOption, nil}if len(opts) != 1 {return nil, errors.New("number of options is more than 1")}opt := opts[0]opt.MagicNumber = DefaultOption.MagicNumberif opt.CodecType == "" {opt.CodecType = DefaultOption.CodecType}return opt, nil
}// Dial connects to an RPC server at the specified network address
func Dial(network, address string, opts ...*Option) (client *Client, err error) {opt, err := parseOptions(opts...)if err != nil {return nil, err}conn, err := net.Dial(network, address)if err != nil {return nil, err}// close the connection if client is nildefer func() {if client == nil {_ = conn.Close()}}()return NewClient(conn, opt)
}
此时，GeeRPC 客户端已经具备了完整的创建连接和接收响应的能力了，最后还需要实现发送请求的能力。
func (client *Client) send(call *Call) {// make sure that the client will send a complete requestclient.sending.Lock()defer client.sending.Unlock()// register this call.seq, err := client.registerCall(call)if err != nil {call.Error = errcall.done()return}// prepare request headerclient.header.ServiceMethod = call.ServiceMethodclient.header.Seq = seqclient.header.Error = ""// encode and send the request// 客户端往连接流中写入数据，及时发送数据给服务端if err := client.cc.Write(&client.header, call.Args); err != nil {call := client.removeCall(seq)// call may be nil, it usually means that Write partially failed,// client has received the response and handledif call != nil {call.Error = errcall.done()}}
}// Go invokes the function asynchronously.
// It returns the Call structure representing the invocation.
func (client *Client) Go(serviceMethod string, args, reply interface{}, done chan *Call) *Call {if done == nil {done = make(chan *Call, 10)} else if cap(done) == 0 {log.Panic("rpc client: done channel is unbuffered")}call := &Call{ServiceMethod: serviceMethod,Args: args,Reply: reply,Done: done,}client.send(call)return call
}// Call invokes the named function, waits for it to complete,
// and returns its error status.
func (client *Client) Call(serviceMethod string, args, reply interface{}) error {call := <-client.Go(serviceMethod, args, reply, make(chan *Call, 1)).Donereturn call.Error
}
Go和Call是客户端暴露给用户的两个RPC服务调用接口，Go是一个异步接口，返回call实例。Call是对Go的封装，阻塞call.Done，等待响应返回，是一个同步接口。
至此，一个支持异步和并发的 GeeRPC 客户端已经完成。
Demo
第一天 GeeRPC 只实现了服务端，因此我们在 main 函数中手动模拟了整个通信过程，今天我们就将 main 函数中通信部分替换为今天的客户端吧。
day2-client/main/main.go
startServer 没有发生变化。
func startServer(addr chan string) {// pick a free portl, err := net.Listen("tcp", ":0")if err != nil {log.Fatal("network error:", err)}log.Println("start rpc server on", l.Addr())addr <- l.Addr().String()geerpc.Accept(l)
}
在 main 函数中使用了 client.Call 并发了 5 个 RPC同步调用，参数和返回值的类型均为 string。
func main() {log.SetFlags(0)addr := make(chan string)go startServer(addr)client, _ := geerpc.Dial("tcp", <-addr)defer func() { _ = client.Close() }()time.Sleep(time.Second)// send request & receive responsevar wg sync.WaitGroupfor i := 0; i < 5; i++ {wg.Add(1)go func(i int) {defer wg.Done()args := fmt.Sprintf("geerpc req %d", i)var reply stringif err := client.Call("Foo.Sum", args, &reply); err != nil {log.Fatal("call Foo.Sum error:", err)}log.Println("reply:", reply)}(i)}wg.Wait()
}
运行结果如下：
start rpc server on [::]:50658
&{Foo.Sum 5 } geerpc req 3
&{Foo.Sum 1 } geerpc req 0
&{Foo.Sum 3 } geerpc req 1
&{Foo.Sum 2 } geerpc req 4
&{Foo.Sum 4 } geerpc req 2
reply: geerpc resp 1
reply: geerpc resp 5
reply: geerpc resp 3
reply: geerpc resp 2
reply: geerpc resp 4
原文地址：https://geektutu.com/post/geerpc-day2.html
