一文搞懂如何實現 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)
}

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

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

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

第一版實現

大家可以先不往下看，自己試着想想該怎麼實現這個函數的超時，第一次嘗試：

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

本文由 Readfog 進行 AMP 轉碼，版權歸原作者所有。
來源：https://mp.weixin.qq.com/s/BERHvjCbxTCEUF8glOBqBg

服務端常見的超時控制

Data race

猜你喜歡