Golang gRPC學習(04): Deadlines超時限制
阿新 • • 發佈：2020-08-15
## 為什麼要使用Deadlines
當我們使用gRPC時，gRPC庫關係的是連線，序列化，反序列化和超時執行。Deadlines 允許gRPC客戶端設定自己等待多長時間來完成rpc操作，直到出現這個錯誤 `DEADLINE_EXCEEDED`。但是在正常情況下，這個DEADLINE_EXCEEDED預設設定是一個很大的數值。
一些語言的API用deadline，一些用 timeout。
在正常情況下，你沒有設定deadline，那麼所有的請求可能在最大請求時間過後才超時。這樣你對於你的伺服器資源，可能存在風險，比如記憶體，可能因為這個長期執行的服務而增長很快，從而耗盡資源。
為了避免這種情況，需要給你的客戶端請求程式設定一個預設的超時時間，在這段時間內請求沒有返回，那麼就超時報錯。
## 怎麼使用？Deadlines使用步驟
### 使用步驟：
**1.設定deadlines**
```go
var deadlineMs = flag.Int("deadline_ms", 20*1000, "Default deadline in milliseconds.")
clientDeadline := time.Now().Add(time.Duration(*deadlineMs) * time.Millisecond)
ctx, cancel := context.WithDeadline(ctx, clientDeadline)
```
**2.檢查deadlines**
```go
if ctx.Err() == context.Canceled {
return status.New(codes.Canceled, "Client cancelled, abandoning.")
}
```
### 具體使用：
**1. 建立連線時超時控制：**
客戶端建立連線時，使用的Dial()函式，它位於
google.golang.org/grpc/clientconn.go 中，我們看看這個函式內容：
```go
func Dial(target string, opts ...DialOption) (*ClientConn, error) {
return DialContext(context.Background(), target, opts...)
}
```
它裡面呼叫的 DialContext() 函式，這個函式非常長，他們在同一個檔案中，它是實際執行的函式，這裡面就有context的timeout和Done相關操作。你也可以到`google.golang.org/grpc/clientconn.go`檔案中去看看這個函式`DialContext`具體是幹嘛的。
使用的時候傳入設定timeout的context，如下：
```go
ctx, cancel := context.Timeout(context.Bakcground(), time.Second*5)
defer cancel()
conn, err := grpc.DialContext(ctx, address, grpc.WithBlock(), grpc.WithInsecure())
```
>- grpc.WithInsecure() ，這個引數啥意思？
>gRPC是建立在HTTP/2上的，所以對TLS提供了很好的支援。如果在客戶端建立連線過程中設定 `grpc.WithInsecure()` 就可以跳過對伺服器證書的驗證。寫練習時可以用這個引數，但是在真實的環境中，不要這樣做，因為有洩露資訊的風險。
>- grpc.WithBlock()
>這個引數會阻塞等待握手成功。
>因為用Dial連線時是非同步連線，連線狀態為正在連線，如果設定了這個引數就是同步連線，會阻塞等待握手成功。
>這個還和超時設定有關，如果你沒有設定這個引數，那麼context超時控制將會失效。
**2. 呼叫時超時：**
函式的呼叫超時控制
```go
ctx, cancel := context.WithTimeout(context.TODO(), time.Second*5)
defer cancel()
result, err := c.SayHello(ctx, &pb.HelloRequest{Name: name})
```
## 例項
用grpc官方的例子來練習下
目錄結構:
```
grpc-tutorial
-- 04deadlines
--client
- main.go
--server
- main.go
--proto/echo
- echo.proto
- echo.pb.go
```
#### 1.首先是定義服務 echo.proto
```bash
syntax = "proto3";
package echo;message
EchoRequest {
string message = 1;
}
message EchoResponse {
string message = 1;
}
service Echo {
rpc UnaryEcho(EchoRequest) returns (EchoRequest) {}
rpc ServerStreamingEcho(EchoRequest) returns (stream EchoResponse) {}
rpc ClientStreamingEcho(stream EchoRequest) returns (EchoResponse) {}
rpc BidirectionalStreamingEcho(stream EchoRequest) returns (stream EchoResponse){}
}
```
進入到proto/echo目錄，生成go檔案，命令如下：
> protoc -I . --go_out=plugins=grpc:. ./echo.proto
#### 2.客戶端
client\main.go
有2個主要的函式，2端都是stream和都不是stream，先看都不是stream的函式
**都不是stream的函式**
```go
// unaryCall 不是stream的請求
func unaryCall(c pb.EchoClient, requestID int, message string, want codes.Code) {
ctx, cancel := context.WithTimeout(context.Background(), time.Second) //超時設定
defer cancel()
req := &pb.EchoRequest{Message: message} //引數部分
_, err := c.UnaryEcho(ctx, req) //呼叫函式傳送請求給服務端
got := status.Code(err) //
fmt.Printf("[%v] wanted = %v, got = %v\n", requestID, want, got)
}
```
上面的code設定在檔案 grpc/codes/codes.go
```go
type Code uint32
const (
OK Code = 0
Canceled Code = 1
Unknown Code = 2
InvalidArgument Code = 3
DeadlineExceeded Code = 4
... ...
)
```
**2端都是stream的函式：**
```go
// streamingCall，2端都是stream
func streamingCall(c pb.EchoClient, requestID int, message string, want codes.Code) {
ctx, cancel := context.WithTimeout(context.Background(), time.Second)//超時設定
defer cancel()
stream, err := c.BidirectionalStreamingEcho(ctx)//雙向stream
if err != nil {
log.Printf("Send error : %v", err)
return
}
err = stream.Send(&pb.EchoRequest{Message: message})//傳送
if err != nil {
log.Printf("Send error : %v", err)
return
}
_, err = stream.Recv() //接收
got := status.Code(err)
fmt.Printf("[%v] wanted = %v, got = %v\n", requestID, want, got)
}
```
**main 執行函式**
```go
func main() {
flag.Parse()
conn, err := grpc.Dial(*addr, grpc.WithInsecure(), grpc.WithBlock())
if err != nil {
log.Fatalf("did not connect : %v ", err)
}
defer conn.Close()
c :=pb.NewEchoClient(conn)
// 成功請求
unaryCall(c, 1, "word", codes.OK)
// 超時 deadline
unaryCall(c, 2, "delay", codes.DeadlineExceeded)
// A successful request with propagated deadline
unaryCall(c, 3, "[propagate me]world", codes.OK)
// Exceeds propagated deadline
unaryCall(c, 4, "[propagate me][propagate me]world", codes.DeadlineExceeded)
// Receives a response from the stream successfully.
streamingCall(c, 5, "[propagate me]world", codes.OK)
// Exceeds propagated deadline before receiving a response
streamingCall(c, 6, "[propagate me][propagate me]world", codes.DeadlineExceeded)
}
```
#### 3.服務端
**定義一個struct**
```go
type server struct {
pb.UnimplementedEchoServer
client pb.EchoClient
cc *grpc.ClientConn
}
```
**2端不是stream的函式：**
```go
func (s *server) UnaryEcho(ctx context.Context, req *pb.EchoRequest)(*pb.EchoResponse, error) {
message := req.Message
if strings.HasPrefix(message, "[propagate me]") {//判斷接收的值
time.Sleep(800 * time.Millisecond)
message := strings.TrimPrefix(message, "[propagate me]")
return s.client.UnaryEcho(ctx, &pb.EchoRequest{Message:message}) //<1>
}
if message == "delay" {
time.Sleep(1500 * time.Millisecond) // message=delay 時睡眠1500毫秒，大於client的設定的1秒，這裡就超時了
}
return &pb.EchoResponse{Message:message}, nil
}
```
上面函式標註 <1> 這個地方比較有意思，當client端傳送的字串包含 [propagate me] 字串時，先睡眠800毫秒，然後在重新執行客戶端請求服務端的函式 s.client.UnaryEcho() , 在次執行到服務端的 UnaryEcho()，客戶端已經超時了。
也就是說client/main.go 先請求了一次服務端，然後在server/main.go 的函式 `func (s *server) UnaryEcho(ctx context.Context, req *pb.EchoRequest) `又執行了一次請求服務端，所以會導致超時。
**2端設定stream的函式：**
```go
func (s *server) BidirectionalStreamingEcho(stream pb.Echo_BidirectionalStreamingEchoServer) error {
for {
req, err := stream.Recv()
if err == io.EOF {
return status.Error(codes.InvalidArgument, "request message not received")
}
if err != nil {
return err
}
message := req.Message
if strings.HasPrefix(message, "[propagate me]") {
time.Sleep(800 * time.Millisecond)
message = strings.TrimPrefix(message, "[propagate me]")
res, err := s.client.UnaryEcho(stream.Context(), &pb.EchoRequest{Message:message})//再次執行客戶端請求服務端函式，這裡可能會超時
if err != nil {
return err
}
stream.Send(res)
}
if message == "delay" {
time.Sleep(1500 * time.Millisecond)
}
stream.Send(&pb.EchoResponse{Message:message})
}
}
```
main函式
```go
func main() {
flag.Parse()
address := fmt.Sprintf(":%v", *port)
lis, err := net.Listen("tcp", address)
if err != nil {
log.Fatalf("failed to listen: %v ", err)
}
echoServer := newEchoServer()
defer echoServer.Close()
grpcServer := grpc.NewServer()
pb.RegisterEchoServer(grpcServer, echoServer)
if err := grpcServer.Serve(lis); err != nil {
log.Fatalf("failed to serve: %v ", err)
}
}
```
#### 執行
先執行 /server/main.go , go run main.go
在執行 /client/main.go, go run main.go
執行結果：
```
go run main.go
[1] wanted = OK, got = OK
[2] wanted = DeadlineExceeded, got = DeadlineExceeded
[3] wanted = OK, got = Unavailable
[4] wanted = DeadlineExceeded, got = Unavailable
[5] wanted = OK, got = Unavailable
[6] wanted = DeadlineExceeded, got = Unavailable
```
## **gRPC 系列程式碼地址：**
>
>- [01hello](https://github.com/jiujuan/grpc-tutorial/tree/master/01hello) grpc helloworld
>- [02fourinteractionmode](https://github.com/jiujuan/grpc-tutorial/tree/master/02fourinteractionmode) grpc 四種傳輸方式
>- [03customer](https://github.com/jiujuan/grpc-tutorial/tree/master/03customer) grpc 一個小練習demo
>- [04deadlines ](https://github.com/jiujuan/grpc-tutorial/tree/master/04deadlines) grpc 超時限制
## 參考：
- [gRPC and Deadlines](https://grpc.io/blog/deadlines/)
- [grpc-go deadline](https://github.com/grpc/grpc-go/tree/master/examples/features/d