golang RPC通信读写超时设置

痛定思痛。 2022-02-01 00:31 493阅读 0赞

golang RPC通信中，有时候就怕读写hang住。

那是否可以设置读写超时呢？

# 1.方案一： 设置连接的读写超时 #

## 1.1 client ##

RPC通信基于底层网络通信，可以通过设置`connection`的读写超时时间，达到RPC读写超时的目的。更多细节可参考[golang网络通信超时设置][golang].

下面以client端的读超时为例，介绍设置方法。

server端和client端代码如下。  
**server**  
一个简单的json RPC server。

package main
    
    import (
        "fmt"
        "log"
        "net"
        "net/rpc"
        "net/rpc/jsonrpc"
        "time"
    )
    
    type Counter struct { 
        Sum int
    }
    
    func (this *Counter) Add(i int, r *int) error { 
        //sleep
        time.Sleep(3*time.Second)
    
        if i < 0 { 
            return fmt.Errorf("data format incorrect")
        }
    
        this.Sum += i
        *r = this.Sum
        log.Printf("i: %v\n", i)
        return nil
    }
    
    func NewJsonRpcSocketServer() { 
    
        rpc.Register(new(Counter))
    
        l, err := net.Listen("tcp", ":3333")
        if err != nil { 
            log.Printf("Listener tcp err: %s", err)
            return
        }
    
        for { 
            log.Println("waiting...")
            conn, err := l.Accept()
            if err != nil { 
                log.Printf("accept connection err: %s\n", conn)
                continue
            }
    
            go jsonrpc.ServeConn(conn)
        }
    
    }
    
    func main() { 
        NewJsonRpcSocketServer()
    }

**client**

json RPC client。

连接建立后，会设置`connection`的读超时时间。

package main
    
    import (
        "log"
        "net"
        "net/rpc/jsonrpc"
        "time"
    )
    
    
    func main() { 
    
         NewJsonRpcSocketClient()
    }
    
    func NewJsonRpcSocketClient() { 
        timeout := time.Second*30
        conn, err := net.DialTimeout("tcp", "127.0.0.1:3333", timeout)
        if err != nil { 
            log.Printf("create client err:%s\n", err)
            return
        }
        defer conn.Close()
    
        // timeout
        readAndWriteTimeout := 2*time.Second
        err = conn.SetDeadline(time.Now().Add(readAndWriteTimeout))
        if err != nil { 
            log.Println("SetDeadline failed:", err)
        }
    
        client := jsonrpc.NewClient(conn)
    
        var reply int
    
        err = client.Call("Counter.Add", 2, &reply)
        if err != nil { 
            log.Println("error:", err, "reply:", reply)
            return
        }
    
        log.Printf("reply: %d, err: %v\n", reply, err)
    
    }

client输出：

2019/05/12 22:52:57 error: read tcp 127.0.0.1:55226->127.0.0.1:3333: i/o timeout reply: 0

## 1.2 server ##

通常情况下，RPC server端的代码如下：

server := rpc.NewServer()
        ... ....
    	for { 
    		conn, err := l.Accept()
    		if err != nil { 
    			log.Println("listener accept fail:", err)
    			time.Sleep(time.Duration(100) * time.Millisecond)
    			continue
    		}
    		
    		// timeout
    		timeout := 10*time.Second
    		conn.SetDeadline(time.Now().Add(timeout))
    		
    		go server.ServeCodec(jsonrpc.NewServerCodec(conn))
    	}

这样，如果设置超时，只有效的影响一次。

对于server来说，都是多次读写。所以，暂时没有方法设置。

# 2.方案二：定时器 #

通过定时器的方式，如果RPC调用在指定时间内没有完成，触发定时器，返回超时错误，关闭连接。

## 2.1 client端 ##

func  RpcCall(method string, args interface{ }, reply interface{ }) error { 
        ... ...
    
    	timeout := time.Duration(10 * time.Second)
    	done := make(chan error, 1)
    
    	go func() { 
    		err := rpcClient.Call(method, args, reply)
    		done <- err
    	}()
    
    	select { 
    	case <-time.After(timeout):
    		log.Printf("[WARN] rpc call timeout %v => %v", rpcClient, RpcServer)
    		rpcClient.Close()
            return fmt.Errorf("timeout")
    	case err := <-done:
    		if err != nil { 
    		    rpcClient.Close()
    			return err
    		}
    	}
    
    	return nil
    }

## 2.2 server端 ##

无论是`gobServerCodec` 或者 `jsonrpc` ，都是实现了`ServerCodec`接口。

`gobServerCodec`文件路径：`/usr/local/go/src/net/rpc/server.go`

`jsonrpc`文件路径：`/usr/local/go/src/net/rpc/server.go`

要给server端RPC读写加上超时机制，需要重新定义`ServerCodec`接口，加上超时的控制。

// A ServerCodec implements reading of RPC requests and writing of
    // RPC responses for the server side of an RPC session.
    // The server calls ReadRequestHeader and ReadRequestBody in pairs
    // to read requests from the connection, and it calls WriteResponse to
    // write a response back. The server calls Close when finished with the
    // connection. ReadRequestBody may be called with a nil
    // argument to force the body of the request to be read and discarded.
    type ServerCodec interface {
    	ReadRequestHeader(*Request) error
    	ReadRequestBody(interface{}) error
    	// WriteResponse must be safe for concurrent use by multiple goroutines.
    	WriteResponse(*Response, interface{}) error
    
    	Close() error
    }

目前，已经有现成的实现方式，可参考[Golang标准库RPC实践及改进][Golang_RPC]

# 3.参考 #

[Does RPC have a timeout mechanism?][Does RPC have a timeout mechanism]

[net/rpc: expected Timeout based alternatives to functions for rpc.Dial, rpc.DialHTTP, rpc.DialHTTPPath \[proposal\]. \#15236][net_rpc_ expected Timeout based alternatives to functions for rpc.Dial_ rpc.DialHTTP_ rpc.DialHTTPPath _proposal_. _15236]

[golang]: https://blog.csdn.net/lanyang123456/article/details/90142129
[Golang_RPC]: http://daizuozhuo.github.io/golang-rpc-practice/
[Does RPC have a timeout mechanism]: https://stackoverflow.com/questions/23330024/does-rpc-have-a-timeout-mechanism
[net_rpc_ expected Timeout based alternatives to functions for rpc.Dial_ rpc.DialHTTP_ rpc.DialHTTPPath _proposal_. _15236]: https://github.com/golang/go/issues/15236