为什么需要超时控制？
　　请求时间过长，用户侧可能已经离开本页面了，服务端还在消耗资源处理，得到的结果没有意义
  ![](http://duan.kim/pig/c.jpg?x-oss-process=style/bb)
　　过长时间的服务端处理会占用过多资源，导致并发能力下降，甚至出现不可用事故
　　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())
　　}

ITPub博客

一文搞懂如何实现 Go 超时控制

最新文章