一文搞懂如何實現Go 超時控制

為什麼需要超時控制?

  • 請求時間過長,用戶側可能已經離開本頁面瞭,服務端還在消耗資源處理,得到的結果沒有意義
  • 過長時間的服務端處理會占用過多資源,導致並發能力下降,甚至出現不可用事故

Go 超時控制必要性

Go 正常都是用來寫後端服務的,一般一個請求是由多個串行或並行的子任務來完成的,每個子任務可能是另外的內部請求,那麼當這個請求超時的時候,我們就需要快速返回,釋放占用的資源,比如goroutine,文件描述符等。

服務端常見的超時控制

  • 進程內的邏輯處理
  • 讀寫客戶端請求,比如HTTP或者RPC請求
  • 調用其它服務端請求,包括調用RPC或者訪問DB等

沒有超時控制會怎樣?

為瞭簡化本文,我們以一個請求函數 hardWork 為例,用來做啥的不重要,顧名思義,可能處理起來比較慢。

func hardWork(job interface{}) error {
  time.Sleep(time.Minute)
  return nil
}

func requestWork(ctx context.Context, job interface{}) error {
 return hardWork(job)
}

這時客戶端看到的就一直是大傢熟悉的畫面

<img src="https://gitee.com/kevwan/static/raw/master/doc/images/loading.jpg" width="25%">

絕大部分用戶都不會看一分鐘菊花,早早棄你而去,空留瞭整個調用鏈路上一堆資源的占用,本文不究其它細節,隻聚焦超時實現。

下面我們看看該怎麼來實現超時,其中會有哪些坑。

第一版實現

大傢可以先不往下看,自己試著想想該怎麼實現這個函數的超時,第一次嘗試:

func requestWork(ctx context.Context, job interface{}) error {
  ctx, cancel := context.WithTimeout(ctx, time.Second*2)
  defer cancel()

  done := make(chan error)
  go func() {
    done <- hardWork(job)
  }()

  select {
  case err := <-done:
    return err
  case <-ctx.Done():
    return ctx.Err()
  }
}

我們寫個 main 函數測試一下

func main() {
  const total = 1000
  var wg sync.WaitGroup
  wg.Add(total)
  now := time.Now()
  for i := 0; i < total; i++ {
    go func() {
      defer wg.Done()
      requestWork(context.Background(), "any")
    }()
  }
  wg.Wait()
  fmt.Println("elapsed:", time.Since(now))
}

跑一下試試效果

➜ go run timeout.go
elapsed: 2.005725931s

超時已經生效。但這樣就搞定瞭嗎?

goroutine 泄露

讓我們在main函數末尾加一行代碼看看執行完有多少goroutine

time.Sleep(time.Minute*2)
fmt.Println("number of goroutines:", runtime.NumGoroutine())

sleep 2分鐘是為瞭等待所有任務結束,然後我們打印一下當前goroutine數量。讓我們執行一下看看結果

➜ go run timeout.go
elapsed: 2.005725931s
number of goroutines: 1001

goroutine泄露瞭,讓我們看看為啥會這樣呢?首先,requestWork 函數在2秒鐘超時後就退出瞭,一旦 requestWork 函數退出,那麼 done channel 就沒有goroutine接收瞭,等到執行 done <- hardWork(job) 這行代碼的時候就會一直卡著寫不進去,導致每個超時的請求都會一直占用掉一個goroutine,這是一個很大的bug,等到資源耗盡的時候整個服務就失去響應瞭。

那麼怎麼fix呢?其實也很簡單,隻要 make chan 的時候把 buffer size 設為1,如下:

done := make(chan error, 1)

這樣就可以讓 done <- hardWork(job) 不管在是否超時都能寫入而不卡住goroutine。此時可能有人會問如果這時寫入一個已經沒goroutine接收的channel會不會有問題,在Go裡面channel不像我們常見的文件描述符一樣,不是必須關閉的,隻是個對象而已,close(channel) 隻是用來告訴接收者沒有東西要寫瞭,沒有其它用途。

改完這一行代碼我們再測試一遍:

➜ go run timeout.go
elapsed: 2.005655146s
number of goroutines: 1

goroutine泄露問題解決瞭!

panic 無法捕獲

讓我們把 hardWork 函數實現改成

panic("oops")

修改 main 函數加上捕獲異常的代碼如下:

go func() {
 defer func() {
  if p := recover(); p != nil {
   fmt.Println("oops, panic")
  }
 }()

 defer wg.Done()
 requestWork(context.Background(), "any")
}()

此時執行一下就會發現panic是無法被捕獲的,原因是因為在 requestWork 內部起的goroutine裡產生的panic其它goroutine無法捕獲。

解決方法是在 requestWork 裡加上 panicChan 來處理,同樣,需要 panicChan 的 buffer size 為1,如下:

func requestWork(ctx context.Context, job interface{}) error {
  ctx, cancel := context.WithTimeout(ctx, time.Second*2)
  defer cancel()

  done := make(chan error, 1)
  panicChan := make(chan interface{}, 1)
  go func() {
    defer func() {
      if p := recover(); p != nil {
        panicChan <- p
      }
    }()

    done <- hardWork(job)
  }()

  select {
  case err := <-done:
    return err
  case p := <-panicChan:
    panic(p)
  case <-ctx.Done():
    return ctx.Err()
  }
}

改完就可以在 requestWork 的調用方處理 panic 瞭。

超時時長一定對嗎?

上面的 requestWork 實現忽略瞭傳入的 ctx 參數,如果 ctx 已有超時設置,我們一定要關註此傳入的超時是不是小於這裡給的2秒,如果小於,就需要用傳入的超時,go-zero/core/contextx 已經提供瞭方法幫我們一行代碼搞定,隻需修改如下:

ctx, cancel := contextx.ShrinkDeadline(ctx, time.Second*2)

Data race

這裡 requestWork 隻是返回瞭一個 error 參數,如果需要返回多個參數,那麼我們就需要註意 data race,此時可以通過鎖來解決,具體實現參考 go-zero/zrpc/internal/serverinterceptors/timeoutinterceptor.go,這裡不做贅述。

完整示例

package main

import (
  "context"
  "fmt"
  "runtime"
  "sync"
  "time"

  "github.com/tal-tech/go-zero/core/contextx"
)

func hardWork(job interface{}) error {
  time.Sleep(time.Second * 10)
  return nil
}

func requestWork(ctx context.Context, job interface{}) error {
  ctx, cancel := contextx.ShrinkDeadline(ctx, time.Second*2)
  defer cancel()

  done := make(chan error, 1)
  panicChan := make(chan interface{}, 1)
  go func() {
    defer func() {
      if p := recover(); p != nil {
        panicChan <- p
      }
    }()

    done <- hardWork(job)
  }()

  select {
  case err := <-done:
    return err
  case p := <-panicChan:
    panic(p)
  case <-ctx.Done():
    return ctx.Err()
  }
}

func main() {
  const total = 10
  var wg sync.WaitGroup
  wg.Add(total)
  now := time.Now()
  for i := 0; i < total; i++ {
    go func() {
      defer func() {
        if p := recover(); p != nil {
          fmt.Println("oops, panic")
        }
      }()

      defer wg.Done()
      requestWork(context.Background(), "any")
    }()
  }
  wg.Wait()
  fmt.Println("elapsed:", time.Since(now))
  time.Sleep(time.Second * 20)
  fmt.Println("number of goroutines:", runtime.NumGoroutine())
}

更多細節

請參考 go-zero 源碼:

  • go-zero/core/fx/timeout.go
  • go-zero/zrpc/internal/clientinterceptors/timeoutinterceptor.go
  • go-zero/zrpc/internal/serverinterceptors/timeoutinterceptor.go

項目地址
https://github.com/tal-tech/go-zero

到此這篇關於一文搞懂如何實現Go 超時控制的文章就介紹到這瞭,更多相關Go 超時控制內容請搜索WalkonNet以前的文章或繼續瀏覽下面的相關文章希望大傢以後多多支持WalkonNet!