Golang标准库RPC实践及改进

转载自：http://daizuozhuo.github.io/golang-rpc-practice/
一直用Golang标准库里的的RPC package来进行远程调用,简单好用. 但是随着任务数量的增大, 发现简单的像包里面的示例那样的代码出现了各种各样的问题,下面就把我踩过的一些坑记录一下吧. 首先是最初使用的文档里的版本,使用HTTP来发送请求.

server.go

func ListenRPC() {rpc.Register(NewWorker())rpc.HandleHTTP()l, e := net.Listen("tcp", ":4200")if e != nil {log.Fatal("Error: listen 4200 error:", e)}go http.Serve(l, nil)
}

client.go

func call(srv string, rpcname string, args interface{}, reply interface{}) error {c, errx := rpc.DialHTTP("tcp", srv+":4200")if errx != nil {return fmt.Errorf("ConnectError: %s", errx.Error())}defer c.Close()return c.Call(rpcname, args, reply)
}

这样四五台机器的情况是够用了, 但是后来集群的机器增加到了十二台, 当请求大了之后发现总有很多任务卡住,通过call函数发送任务之后总会有没有返回的情况. 于是转而直接用tcp,效率有很大提升.

server.go

func ListenRPC() {rpc.Register(NewWorker())l, e := net.Listen("tcp", ":4200")if e != nil {log.Fatal("Error: listen 4200 error:", e)}go func() {for {conn, err := l.Accept()if err != nil {log.Print("Error: accept rpc connection", err.Error())continue}go rpc.ServeConn(conn)}}()
}

client.go

func call(srv string, rpcname string, args interface{}, reply interface{}) error {c, errx := rpc.Dial("tcp", srv+":4200")if errx != nil {return fmt.Errorf("ConnectError: %s", errx.Error())}defer c.Close()return c.Call(rpcname, args, reply)
}

这样局面有所改观,但是还是有任务卡住,概率大概是0.01%, 也就是一万个call里会有一个没有响应. 仔细研究后发现这个rpc package有两大坑:

rpc包里的rpc.Dial函数没有timeout, 系统默认是没有timeout的,所以在这里可能卡住.所以我们可以采用net包里的 net.DialTimeout函数.

rpc包里默认使用gobCodec来编码解码, 这里io可能会卡住而不返回错误,所以我们要自己编写加入timeout的codec. 注意server这边读写都有timeout,但是client这边只有写有timeout,因为读的话并不能预知任务完成的时间. 于是就有了接下来这个版本的rpc,几十万个任务下来没有任何问题.

完整的代码可以在在github rpc-example上下载.

server.go

func TimeoutCoder(f func(interface{}) error, e interface{}, msg string) error {echan := make(chan error, 1)go func() { echan <- f(e) }()select {case e := <-echan:return ecase <-time.After(time.Minute):return fmt.Errorf("Timeout %s", msg)}
}type gobServerCodec struct {rwc    io.ReadWriteCloserdec    *gob.Decoderenc    *gob.EncoderencBuf *bufio.Writerclosed bool
}func (c *gobServerCodec) ReadRequestHeader(r *rpc.Request) error {return TimeoutCoder(c.dec.Decode, r, "server read request header")
}func (c *gobServerCodec) ReadRequestBody(body interface{}) error {return TimeoutCoder(c.dec.Decode, body, "server read request body")
}func (c *gobServerCodec) WriteResponse(r *rpc.Response, body interface{}) (err error) {if err = TimeoutCoder(c.enc.Encode, r, "server write response"); err != nil {if c.encBuf.Flush() == nil {log.Println("rpc: gob error encoding response:", err)c.Close()}return}if err = TimeoutCoder(c.enc.Encode, body, "server write response body"); err != nil {if c.encBuf.Flush() == nil {log.Println("rpc: gob error encoding body:", err)c.Close()}return}return c.encBuf.Flush()
}func (c *gobServerCodec) Close() error {if c.closed {// Only call c.rwc.Close once; otherwise the semantics are undefined.return nil}c.closed = truereturn c.rwc.Close()
}func ListenRPC() {rpc.Register(NewWorker())l, e := net.Listen("tcp", ":4200")if e != nil {log.Fatal("Error: listen 4200 error:", e)}go func() {for {conn, err := l.Accept()if err != nil {log.Print("Error: accept rpc connection", err.Error())continue}go func(conn net.Conn) {buf := bufio.NewWriter(conn)srv := &gobServerCodec{rwc:    conn,dec:    gob.NewDecoder(conn),enc:    gob.NewEncoder(buf),encBuf: buf,}err = rpc.ServeRequest(srv)if err != nil {log.Print("Error: server rpc request", err.Error())}srv.Close()}(conn)}}()
}

client.go

type gobClientCodec struct {rwc    io.ReadWriteCloserdec    *gob.Decoderenc    *gob.EncoderencBuf *bufio.Writer
}func (c *gobClientCodec) WriteRequest(r *rpc.Request, body interface{}) (err error) {if err = TimeoutCoder(c.enc.Encode, r, "client write request"); err != nil {return}if err = TimeoutCoder(c.enc.Encode, body, "client write request body"); err != nil {return}return c.encBuf.Flush()
}func (c *gobClientCodec) ReadResponseHeader(r *rpc.Response) error {return c.dec.Decode(r)
}func (c *gobClientCodec) ReadResponseBody(body interface{}) error {return c.dec.Decode(body)
}func (c *gobClientCodec) Close() error {return c.rwc.Close()
}func call(srv string, rpcname string, args interface{}, reply interface{}) error {conn, err := net.DialTimeout("tcp", srv+":4200", time.Second*10)if err != nil {return fmt.Errorf("ConnectError: %s", err.Error())}encBuf := bufio.NewWriter(conn)codec := &gobClientCodec{conn, gob.NewDecoder(conn), gob.NewEncoder(encBuf), encBuf}c := rpc.NewClientWithCodec(codec)err = c.Call(rpcname, args, reply)errc := c.Close()if err != nil && errc != nil {return fmt.Errorf("%s %s", err, errc)}if err != nil {return err} else {return errc}
}

Golang标准库RPC实践及改进相关推荐

Golang 标准库log的实现
原创作品,允许转载,转载时请务必以超链接形式标明文章原始出处 .作者信息和本声明.否则将追究法律责任.http://gotaly.blog.51cto.com/8861157/1406905 前一篇 ...
golang 标准库间依赖的可视化展示
简介国庆看完 << Go 语言圣经 >>,总想做点什么,来加深下印象.以可视化的方式展示 golang 标准库之间的依赖,可能是一个比较好的切入点.做之前,简单搜了下相关的内 ...
Golang标准库CHM格式文档
上手Go后,想熟悉golang标准库来做一些项目.在学习和使用golang标准库的时候,发现golang标准库文档不太友好.主要是导航区域和内容区域无法同屏浏览,在包和包间.包内不同对象间来回切换 ...
golang标准库os模块-文件目录相关
golang标准库os模块-文件目录相关本文视频教程:https://www.bilibili.com/video/BV1zR4y1t7Wj?from=search&seid=7990946 ...
Golang标准库-syscall（什么是系统调用/Go 语言中的系统调用）
文章目录一.什么是系统调用二.Golang标准库-syscall 1. syscall无处不在 2. syscall demo举例: go版本的strace Strace go版本的strace ...
Golang标准库中的fmt
Golang标准库中的fmt fmt包实现了类似C语言printf和scanf的格式化I/O.主要分为向外输出内容和获取输入内容两大部分. 1. 向外输出标准库fmt提供了以下几种输出相关函数. P ...
golang标准库http服务器处理流程
http标准库 golang本身就提供了http的标志库,在golang中可以轻松的编写http服务,本文主要是因为在编写http服务的过程中,对整个处理流程不是很了解故想了解一下. 示例代码 pac ...
Golang 标准库提供的Log(一)
原创作品,允许转载,转载时请务必以超链接形式标明文章原始出处 .作者信息和本声明.否则将追究法律责任.http://gotaly.blog.51cto.com/8861157/1405754 Gol ...
Go语言自学系列 | golang标准库os模块 - 文件目录相关
视频来源:B站<golang入门到项目实战 [2021最新Go语言教程,没有废话,纯干货!持续更新中...]> 一边学习一边整理老师的课程内容及试验笔记,并与大家分享,请移步至知乎网站,谢 ...

Golang标准库RPC实践及改进

Golang标准库RPC实践及改进相关推荐

最新文章

热门文章