一、概述

在Rpcx框架源码中，存在Client的角色，用来完成承担Client stub；相对来说Server，通过从注册中心获取已注册暴露对外提供功能的Service，本将已注册的service存放到本地，减少频繁的读取registry中心的记录，同时也通过watch机制获取registry中注册服务的变更(采用channel来监听service的变更)，来同步本地缓存的记录，通过RPCClient与server完成通信采用同步或异步的通信方式。对应的Clinet提供了如下的功能：

func (client *Client) Call(ctx context.Context, servicePath, serviceMethod string,
                        args interface{}, reply interface{}) error // 同步调用
func (client *Client) Close() error // 释放RPCClient相关资源
func (c *Client) Connect(network, address string) error // 连接server
func (client *Client) Go(ctx context.Context, servicePath, serviceMethod string,
           args interface{}, reply interface{}, done chan *Call) *Call // 异步
func (client *Client) IsClosing() bool  // 本地RPCClient是否已关闭
func (client *Client) IsShutdown() bool // RPCClient是否停止

func (client *Client) SendRaw(ctx context.Context, r *protocol.Message) 
          (map[string]string, []byte, error) //原始信息发送

对应的XClinet提供了如下的功能：(XClient是在Client的基础上增加路由规则、失败模式、服务发现治理等功能)

func (c *xClient) SetSelector(s Selector)  // 设置路由规则
func (c *xClient) SetPlugins(plugins PluginContainer) // 设置插件 增强client特性
func (c *xClient) ConfigGeoSelector(latitude, longitude float64) // 就近路由规则
func (client *Client) Call(ctx context.Context, servicePath,serviceMethod string, 
      args interface{},reply interface{}) error // 同步调用
func (client *Client) Go(ctx context.Context, servicePath,serviceMethod string, 
      args interface{}, reply interface{}, done chan *Call) *Call // 异步执行调用
func (client *Client) SendRaw(ctx context.Context, r *protocol.Message) 
         (map[string]string, []byte, error)  // 发送原始信息
func (c *xClient) Broadcast(ctx context.Context, serviceMethod string, 
      args interface{},  reply interface{}) error  // 广播请求模式
func (c *xClient) Fork(ctx context.Context, serviceMethod string, 
    args interface{},  reply interface{}) error // Fock 模式请求

接下来详细剖析对应的源码

二、源码

本文源码主要位于：rpcx\client包中client.go 和 xclient.go
Client结构

// 支持同步、异步的调用模式；一个Client就代表一个RPCClient
type Client struct {
    option Option // 配置选项

    Conn net.Conn      // connection
    r    *bufio.Reader   // 读

    mutex        sync.Mutex // protects following
    seq          uint64
    pending      map[uint64]*Call    // 挂起的调用
    closing      bool      // user has called Close  client关闭
    shutdown     bool    // server has told us to stop  client停止
    pluginClosed bool  // the plugin has been called plugin禁用

    Plugins PluginContainer         // 插件容器

    ServerMessageChan chan<- *protocol.Message  // 接收server的通知channel
}

从上面的Client结构体的源码可以看到提供了Client只是具备了简单RPC通信功能：同步和异步，并以Plugin方式提供Client的额外特性、Client权限凭证等操作；具体的提供的函数如下：

1.1 新建Client

// 新建Client
// 目前支持Option：服务分组、重试、TLS配置、连接timeout、读取timeout、
// 写入timeout、调用备份失败有效期、断路器、协议序列化、协议压缩方式、心跳等相关参数设置
// 默认已提供DefaultOption满足基本参数设置需求，可以在此基础上按需修改.
func NewClient(option Option) *Client {
    return &Client{
        option: option,
    }
}

1.2 连接Server

// 连接server并准备发送client的请求：client是需要指定的network来连接server的
// 本部分源码位于：client包下的connection.go文件中
// 
func (c *Client) Connect(network, address string) error {
    var conn net.Conn
    var err error

    switch network {
    case "http":   
        conn, err = newDirectHTTPConn(c, network, address)
    case "kcp":
        conn, err = newDirectKCPConn(c, network, address)
    case "quic":
        conn, err = newDirectQuicConn(c, network, address)
    case "unix":
        conn, err = newDirectConn(c, network, address)
    default:  // 默认TCP
        fn := makeConnMap[network]
        if fn != nil {   
            conn, err = fn(c, network, address)
        } else {  // TLS方式连接 
            conn, err = newDirectConn(c, network, address)
        }
    }

    if err == nil && conn != nil {  // 设置connection的client端相关参数
        if c.option.ReadTimeout != 0 { // 读超时
            conn.SetReadDeadline(time.Now().Add(c.option.ReadTimeout))
        }
        if c.option.WriteTimeout != 0 { // 写超时
            conn.SetWriteDeadline(time.Now().Add(c.option.WriteTimeout))
        }

        c.Conn = conn
        c.r = bufio.NewReaderSize(conn, ReaderBuffsize)  // 设置buffer
        //c.w = bufio.NewWriterSize(conn, WriterBuffsize)

        // start reading and writing since connected
        go c.input()   // 等待connected后 执行read或write操作

                // 额外开启goroutine来执行heartbeat
        if c.option.Heartbeat && c.option.HeartbeatInterval > 0 { 
            go c.heartbeat()
        }

    }

    return err
}

等待连接正常进行read或write操作的入口

// 读取server的响应结果：
// 1、本地缓存不存在对应调用对象call记录
// 2、调用对象响应结果返回：a、出现了Error b、正常
func (client *Client) input() { //
    var err error
    var res = protocol.NewMessage() // 请求消息

        // for循环根据用户定义的内容来构建成发送给server端的protocol.Message信息
       //              读取server返回给客户端的protocol.Message信息
       // client相关的timeou参数设置
    for err == nil {
        if client.option.ReadTimeout != 0 { // 设置读取有效期
            client.Conn.SetReadDeadline(time.Now().Add(client.option.ReadTimeout))
        }

        // 完成将用户定义的request转为Message
        err = res.Decode(client.r) // 解码
        //res, err = protocol.Read(client.r)

        if err != nil {
            break
        }
        // 验证Message的合法性
        seq := res.Seq()
        var call *Call
        isServerMessage := (res.MessageType() == protocol.Request 
                   && !res.IsHeartbeat() && res.IsOneway()) // 是否为server消息
// 当前Message属于request的时  获取其关联的pending中对应call：
//     pending中记录的是待执行的call
        if !isServerMessage {  
            client.mutex.Lock()
            call = client.pending[seq]
            delete(client.pending, seq)
            client.mutex.Unlock()
        }

        switch {
        case call == nil: // 当调用对象call不存在本地缓存中
            if isServerMessage {  // 
                if client.ServerMessageChan != nil {
                                       // 新建go协程单独处理response结果
                    go client.handleServerRequest(res) 
                    res = protocol.NewMessage()
                }
                continue
            }
        case res.MessageStatusType() == protocol.Error: // 消息状态 = Error
            // We've got an error response. Give this to the request
            // 当server端输出一个错误响应 给到对应的请求
            if len(res.Metadata) > 0 { // 提取响应内容中meta data数据
                meta := make(map[string]string, len(res.Metadata))
                for k, v := range res.Metadata {
                    meta[k] = v
                }
                call.ResMetadata = meta
                call.Error = ServiceError(meta[protocol.ServiceError])
            }

            if call.Raw { // 调用使用的raw形式  需要将响应内容进行转换
                call.Metadata, call.Reply, _ = convertRes2Raw(res)
                call.Metadata[XErrorMessage] = call.Error.Error()
            }
            call.done() // 触发结果通道channel
        default: // 响应处理
            if call.Raw { // Raw形式单独提取出来 完成对应的转换：获取metadata、reply
                call.Metadata, call.Reply, _ = convertRes2Raw(res)
            } else {
                data := res.Payload // 提取响应体
                if len(data) > 0 {
                                       // 根据对应的解码来进行解码
                    codec := share.Codecs[res.SerializeType()] 
                    if codec == nil {
                        call.Error = ServiceError(ErrUnsupportedCodec.Error())
                    } else {
                  // 将对应的返回结果赋给本次rpc的Call中的Reply
                        err = codec.Decode(data, call.Reply) 
                        if err != nil {
                            call.Error = ServiceError(err.Error())
                        }
                    }
                }
                if len(res.Metadata) > 0 { // 提取metadata
                    meta := make(map[string]string, len(res.Metadata))
                    for k, v := range res.Metadata {
                        meta[k] = v
                    }
// 赋予本次call对应的响应元数据ResMetadata
                    call.ResMetadata = res.Metadata 
                }

            }

            call.done() // 通知结果通道
        }

        res.Reset() // 重置Message对象 便于下一次新的请求Message处理
    }

    ...省略代码
}

每次进行的RPC过程会被封装成Call对象，通过Done的channel来监控每次Call完成情况

// Call represents an active RPC.
type Call struct {
    ServicePath   string               // RPC调用的Service名字及其对应的Method.
    ServiceMethod string             // 
    Metadata      map[string]string  //metadata
    ResMetadata   map[string]string  // 响应结果的metadata
    Args          interface{} // 函数的参数(*struct).
    Reply         interface{} // 函数的应答结果 (*struct).
    Error         error       // 一次Call完成时对应的Error状态
    Done          chan *Call  // 存放每次完成的Call.
    Raw           bool        // 是否采用原始数据信息方式
}

发起远程调用（同步调用）

func (client *Client) call(ctx context.Context, servicePath, serviceMethod string,
     args interface{}, reply interface{}) error {
        seq := new(uint64)                          // 请求ID
     // 将seq添加到context中便于在Server端时获取
    ctx = context.WithValue(ctx, seqKey{}, seq) 
    Done := client.Go(ctx, servicePath, serviceMethod, args, 
            reply, make(chan *Call, 1)).Done
        // 异步请求
    var err error
    select {
    case <-ctx.Done(): //cancel by context  取消
        client.mutex.Lock()  
    // 使用互斥锁 剔除处于pending状态的call； 
    //  处于pending状态的call存放在client端的本地内存中map[uint64]*Call
        call := client.pending[*seq]
        delete(client.pending, *seq)
        client.mutex.Unlock()
        if call != nil { //  将本次Call投递到channel中
            call.Error = ctx.Err()
            call.done() // 执行Call投递到channel
        }

        return ctx.Err() // 输出本次request的错误状态
    case call := <-Done: // 从Done的channel获取Call：代表Call已完成
        err = call.Error
        meta := ctx.Value(share.ResMetaDataKey) // 获取response内容
        if meta != nil && len(call.ResMetadata) > 0 {
            resMeta := meta.(map[string]string)
            for k, v := range call.ResMetadata {
                resMeta[k] = v
            }
        }
    }

    return err

目前rpcx 支持许多服务发现机制，同时也可以自定义服务发现：

• [Peer to Peer] 客户端直连每个服务节点。
• [Peer to Multiple]客户端可以连接多个服务。服务可以被编程式配置。
• [Zookeeper] 通过 zookeeper 寻找服务。
• [Etcd] 通过 etcd 寻找服务。
• [Consul] 通过 consul 寻找服务。
• [mDNS] 通过 mDNS 寻找服务（支持本地服务发现）。
• [In process]在同一进程寻找服务。客户端通过进程调用服务，不走TCP或UDP，方便调试使用。

rpcx 支持 故障模式：

• [Failfast]：如果调用失败，立即返回错误
• [Failover]：选择其他节点，直到达到最大重试次数
• [Failtry]：选择相同节点并重试，直到达到最大重试次数
• [Failbackup]：选择两个节点，当一个节点未能返回结果时，则请求第二个节点来获取结果

rpcx 支持 路由模式：

[Random]： 随机选择节点
[Roundrobin]： 用 [roundrobin] (https://zh.wikipedia.org/wiki/%E5%BE%AA%E7%92%B0%E5%88%B6) 算法选择节点
[Consistent hashing]: 如果服务路径、方法和参数一致，就选择同一个节点， 使用了非常快的jump consistent hash算法
[Weighted]: 根据元数据里配置好的权重(weight=xxx)来选择节点
[Network quality]: 根据ping的结果来选择节点。网络质量越好，该节点被选择的几率越大
[Geography]: 如果有多个数据中心，客户端趋向于连接同一个数据机房的节点
[Customized Selector]: 如果以上的选择器都不适合你，你可以自己定制选择器

rpcx特殊的两种请求方式

• Broadcast 表示向所有服务器发送请求，只有所有服务器正确返回时才会成功。此时FailMode 和 SelectMode的设置是无效的。请设置超时来避免阻塞。
• Fork 表示向所有服务器发送请求，只要任意一台服务器正确返回就成功。此时FailMode 和 SelectMode的设置是无效的。

三、使用

实例server端代码

import (
    "context"
    "flag"
    "github.com/smallnest/rpcx/server"
    "log"
    "net"
    "rpcx/examples/models"
    )

var (
    addr = flag.String("addr","localhost:8972","server address")
)

func main() {
    flag.Parse()

    s := server.NewServer()
    //s.Plugins.Add(&ConnectionListerPlugin{})

    s.Register(new(models.Arith),"")  // 只提供rcvr不指定servicePath和method以及对应的name
    s.RegisterName("PBArith", new(models.PBArith),"") // 提供rcvr及其name
    s.RegisterFunction("PB-Mul",models.Mul,"") // 注入函数(对应的函数不需要提供调用方) 提供servicePath和method
    s.RegisterFunctionName("PMul","mul",models.Mul,"") // 提供servicePath和method及其name

    s.Serve("tcp", *addr)
    //log.Println(" Server Address= " + s.Address().String())
}

type ConnectionListerPlugin struct {
}

func (clis *ConnectionListerPlugin) HandleLister(conn net.Conn) (net.Conn, bool){
    log.Printf("Server Listener Address %v \n", conn.LocalAddr().String())
    return conn,true
}

func mul(ctx context.Context, args *models.Args, reply *models.Reply) error {
    reply.C = args.A * args.B
    return nil
}

实例client端代码

package main

import (
    "context"
    "flag"
    "rpcx/client"
    "rpcx/examples/models"
    "log"
    "time"
)

var  (
    addr = flag.String("addr","localhost:8972","server address")
)

func main() {
    flag.Parse()
    //
    d := client.NewPeer2PeerDiscovery("tcp@"+*addr, "")
    xclient := client.NewXClient("Arith", client.Failtry, client.RandomSelect, d, client.DefaultOption)
    defer xclient.Close()

    args := &models.Args{
        A: 10,
        B: 20,
    }

    for{
        reply := &models.Reply{}
        err := xclient.Call(context.Background(),"Mul", args, reply)
        if err != nil{
            log.Fatalf("failed to call:%v", err)
        }
        log.Printf("%d * %d = %d", args.A, args.B, reply.C)

        time.Sleep(1e9)
    }
}

辅助代码

package models

import (
    "context"
    "fmt"
    "rpcx/_testutils"
)

// 参数
type Args struct {
    A int
    B int
}

// 回复结果
type Reply struct {
    C int
}

// 服务
type Arith int

func (t *Arith) Mul(ctx context.Context, args *Args, reply *Reply) error {
    reply.C = args.A * args.B
    fmt.Printf("call: %d * %d = %d\n", args.A, args.B, reply.C)
    return nil
}

func (t *Arith) Add(ctx context.Context, args *Args, reply *Reply) error {
    reply.C = args.A + args.B
    fmt.Printf("call: %d + %d = %d\n", args.A, args.B, reply.C)
    return nil
}

func (t *Arith) Say(ctx context.Context, args *string, reply *string) error {
    *reply = "hello " + *args
    return nil
}

type PBArith int

func (t *PBArith) Mul(ctx context.Context, args *testutils.ProtoArgs, reply *testutils.ProtoReply) error {
    reply.C = args.A * args.B
    return nil
}

func (t *Arith) ThriftMul(ctx context.Context, args *testutils.ThriftArgs_, reply *testutils.ThriftReply) error {
    reply.C = args.A * args.B
    return nil
}

func Mul(ctx context.Context, args *testutils.ProtoArgs, reply *testutils.ProtoReply) error {
    reply.C = args.A * args.B
    return nil
}

四、其他

Client分析源码
Client执行链路如下：