agent通过RCP向transfer发送metrics时,应用rpc长连贯,通过net/rpc保护一个tcp connection。
agent封装了SingleConnRpcClient构造,来实现单连贯rcp client。
1. SingleConnRpcClient如何应用的?
agent对应的后端transfer有多个,只有向1个transfer发送胜利就OK;
发送时,先依据addr找可用的rpcclient,找不到就init一个,而后调用update实现调用操作。
func SendMetrics(metrics []*model.MetricValue, resp *model.TransferResponse) {
rand.Seed(time.Now().UnixNano())
for _, i := range rand.Perm(len(Config().Transfer.Addrs)) { //把数组的程序打乱
addr := Config().Transfer.Addrs[i]
c := getTransferClient(addr) //从map中找
if c == nil {
c = initTransferClient(addr) //没找到,进行初始化
}
if updateMetrics(c, metrics, resp) { //发送,若胜利则break
break
}
}
}
初始化rpcclient:
func initTransferClient(addr string) *SingleConnRpcClient {
var c *SingleConnRpcClient = &SingleConnRpcClient{
RpcServer: addr,
Timeout: time.Duration(Config().Transfer.Timeout) * time.Millisecond,
}
......
TransferClients[addr] = c
return c
}
发送的逻辑:调用RPC办法:Transfer.Update
func updateMetrics(c *SingleConnRpcClient, metrics []*model.MetricValue, resp *model.TransferResponse) bool {
err := c.Call("Transfer.Update", metrics, resp)
if err != nil {
log.Errorln("failed to call Transfer.Update:", c, err)
return false
}
return true
}
2. SingleConnRpcClient如何实现的?
先看一下数据结构:
type SingleConnRpcClient struct {
sync.Mutex
rpcClient *rpc.Client
RpcServer string
Timeout time.Duration
}
为防止并发调用,外部封装了sync.Mutext;
重点是Call()办法:
- 先看一下连贯在不在,不在的话要init一个;
- 启动一个goroutine来执行rpc办法;
- 在主流程中用select进行超时管制,只有超时就close连贯;
func (this *SingleConnRpcClient) Call(method string, args interface{}, reply interface{}) error {
this.Lock()
defer this.Unlock()
err := this.serverConn()
if err != nil {
return err
}
timeout := time.Duration(10 * time.Second)
done := make(chan error, 1)
go func() {
err := this.rpcClient.Call(method, args, reply)
done <- err
}()
select {
case <-time.After(timeout):
log.Printf("[WARN] rpc call timeout %v => %v", this.rpcClient, this.RpcServer)
this.close()
case err := <-done: //失常返回err=nil
if err != nil {
log.Printf("[ERROR] rpc call with error => %v -- %v", err, this.rpcClient)
this.close()
return err
}
}
return nil
}
初始化连贯很简略,就是创立了一个JsonRpCClient:
func (this *SingleConnRpcClient) serverConn() error {
if this.rpcClient != nil {
return nil
}
var err error
var retry int = 1
for {
if this.rpcClient != nil {
return nil
}
this.rpcClient, err = net.JsonRpcClient("tcp", this.RpcServer, this.Timeout)
if err != nil {
log.Printf("dial %s fail: %v", this.RpcServer, err)
if retry > 3 {
return err
}
time.Sleep(time.Duration(math.Pow(2.0, float64(retry))) * time.Second)
retry++
continue
}
return err
}
}
连贯敞开:
func (this *SingleConnRpcClient) close() {
if this.rpcClient != nil {
this.rpcClient.Close()
this.rpcClient = nil
}
}