Go context

JunChow520 · · 446 次点击 · · 开始浏览    
这是一个创建于 的文章,其中的信息可能已经有所发展或是发生改变。

控制并发有两种经典的方式:WaitGroup和Context

  • WaitGroup:控制多个Goroutine同时完成
  • Context:并发控制和超时控制的标准做法

WaitGroup

WaitGroup控制多个Goroutine同时完成,适用于多个Goroutine协同完成一项任务时,由于每个Goroutine做的都是整体的一部分,只有全部Goroutine都完成,整个任务才算完成,因此会采用等待组的方式。

var wg sync.WaitGroup
wg.Add(2)

go func(){
    defer wg.Done()
    time.Sleep(time.Second * 2)
    fmt.Printf("PART1: Finished\n")
}()
go func(){
    defer wg.Done()
    time.Sleep(time.Second * 2)
    fmt.Printf("PART2: Finished\n")
}()

wg.Wait()
fmt.Printf("Task finished\n")

WaitGroup只会傻傻地等待子Goroutine结束,并不能主动通知子Goroutine退出。

Channel + Select

实际业务中存在这样的场景:需要主动通知某个Goroutine结束,比如开启一个后台Goroutine用于监控,现在不需要了就需要通知这个监控的Goroutine结束,不然它会一直运行出现泄露。

当一个Goroutine启动后是无法控制的,大部分情况只能等待它自己结束,若Goroutine是一个不会自己结束的后台Goroutine,比如监控它会一直运行的。对于这种情况,最傻瓜的方式是采用全局变量,在其它位置修改全局变量完成结束通知,然后在这个Goroutine中不停的检查这个变量。若发现被通知关闭了就自我结束。虽然这种方式可行,但首先需要确保这个变量在多线程下的安全,基于此有了更好的方式channel+select

//stop用于通知结束后台Goroutine
stop := make(chan bool)
//后台Goroutine
go func() {
    for {
        //select判断stop是否接收到值,若接收到则退出停止,若无则执行default中的监控逻辑。
        select {
        case <-stop:
            fmt.Printf("monitor quit\n")
            return
        default:
            fmt.Printf("monitor running...\n")
            time.Sleep(time.Second * 1)
        }
    }
}()

time.Sleep(time.Second * 5)
fmt.Printf("notificate monitor stop...\n")
stop <- true
time.Sleep(time.Second * 5)

channel + select的方式比较优雅的结束了一个Goroutine,不存也有局限性。

Golang中不能直接杀死Goroutine,Goroutine的关闭一般采用channel + select的方式来控制。但在某些场景下,比如处理一个请求衍生了很多相互关联的Goroutine,它们共享一些全局变量或存在共同的deadline等,可以同时被关闭,使用channel + select会比较麻烦,此时就可以通过Context来实现。

比如:一个网络请求,每个请求都需要开启一个Goroutine做一些事情,这些Goroutine又可能会开启其它的Goroutine。因此需要一种可以跟踪Goroutine的方案,才可以达到控制他们的目的,这就是Golang提供的Context。

parent := context.Background()
ctx,cancel := context.WithCancel(parent)

go func(ctx context.Context){
    for{
        select{
        case <-ctx.Done():
            fmt.Printf("monitor quit...\n")
            return
        default:
            fmt.Printf("monitor running...\n")
            time.Sleep(time.Second * 1)
        }
    }
}(ctx)

time.Sleep(time.Second * 3)
fmt.Printf("notification monitor stop\n")
cancel()
time.Sleep(time.Second * 3)

Context

Golang的设计者考虑到多个Goroutine共享数据及其管理而设计了context包,Golang标准库提供了context包通过上下文管理子协程的生命周期。

例如:在Go的服务端中,通常每个请求都会开启若干个Goroutine同时工作,服务端实际上就是一个协程模型。

协程模型

Context可理解为程序单元的一个运行状态、现场、快照,Golang中程序单元也就是Goroutine。因此准确来说是Goroutine的上下文。

Goroutine是一个轻量级的执行线程,多个Goroutine要比一个线程轻量,因此管理多个Goroutine消耗的资源相对会更少。每个Goroutine执行前都需知道当前的执行状态,上下文会将执行状态封装在一个Context变量中,传递给要执行的Goroutine中。

Context携带着截止时间、cancel信号、还有在API之间提供值读写,用于在Goroutine之间传递信息。

Context主要用于父子任务之间的同步取消信号,本质上是一种Goroutine调度的方式。

Context是线程安全的,因为context本身是不可变的(immutable),因此可以放心地在多个Goroutine中传递使用。

context包通过构建树型关系的Context来达到上一层Goroutine能够对传递给下一层Goroutine的控制,对于处理一个请求操作,需要采用context来层层控制Goroutine以及传递变量来共享。

  • Context对象的生存周期一般仅为一个请求的处理周期,针对一个请求创建一个Context变量作为Context跟结构的根节点,当请求处理结束后需撤销此ctx变量以释放资源。
  • 每次创建一个Goroutine,要么将原有的Context传递给Goroutine,要么创建一个子Context并传递给Goroutine。
  • Context能够灵活地存储不同类型、不同数量的值,并使多个Goroutine安全地读写其中的值。
  • 当通过父Context对象创建子Context时,可同时获得子Context的撤销函数,这样父Context对象的创建环境就获得了对子Context将要传递到的Goroutine的撤销权。

使用原则

使用Context的程序包需要遵循如下原则来满足接口一致性以便于静态分析

  • 不要将Context保存在一个Struct当中,显式地传入参数。Context变量需要作为第一个参数使用,一般命名为ctx
  • 即使方法允许也不要传入一个nil的Context,若不确定要用什么Context的时候可传入一个context.TDOO
  • Context可用来传递到不同的Goroutine中,Context在多个Goroutine中是安全的。

在子Context被传递到的Goroutine中应对该子Context的Done信道进行监控,一旦该信号被关闭即上层运行环境撤销了本Goroutine的执行,应主动终止对当前请求信息的处理,同时释放资源并返回。

退出通知

  • Context很好的解决了多个Goroutine通知传递和元数据问题

由于Golang中Goroutine之间没有父子关系,因此也不存在子进程退出后的通知机制。多个Goroutine协同工作主要涉及四个方面:

  • 通信:Channel是多Goroutine之间通信的基础
  • 同步:不带缓冲的Channel、WaitGroup为多Goroutine提供同步等待机制,Mutex互斥锁和读写锁机制。
  • 通知:通知和上下文通信的区别在于,通知的作用是管理控制流数据,常用的方式是在输入端绑定两个Channel,通过Select收敛处理。Select+Channel可解决简单问题,但不是一个通用的解决方案。
  • 退出:简单的解决方案和通知类似,增加一个单独的Channel,借助Channnel和Select的广播机制实现退出。

由于Goroutine之间是平等的,当遇到复杂的并发结构时处理退出机制则会显得力不从心。

  • Context几乎已经成为传递与请求同生存周期变量的标准方法

例如:网络编程中当接收到一个请求处理时可能需要开启不同的Goroutine来获取数据并处理逻辑,因此存在一个请求会在多个Goroutine中处理的情况,而这些Goroutine可能会共享请求的一些信息。同时当请求被取消或超时,所有从这个请求创建的Goroutine也应该被结束。

Context

如此一来通过传递Context就可以追踪Goroutine调用树,并在这些调用树之间传递通知和元数据。虽然Goroutine之间是平行的,没有继承关系,但Context设计成是包含父子关系的,这样可以更好地描述Goroutine调用之间的树型关系。

context.Context

context包的核心是Context接口

type Context interface {
    Deadline() (deadline time.Time, ok bool)
    Done() <-chan struct{}
    Err() error
    Value(key interface{}) interface{}
}

Context接口提供了4个方法

  • DeadLine方法用来一个超时时间,当Goroutine获取超时时间后可对io操作设置超时。
  • Done方法用来获取用户一个信道,当上下文被撤销或过期时会关闭信道,以此作为上下文是否已经关闭的信号。
  • Err方法当Done操作信道关闭后,Err方法用来表明信道被撤销的原因。
  • Value方法用于让Goroutine共享数据

context.Context类型的值可以协调多个Goroutine中代码执行【取消】操作,并可存储键值对。最重要的是它是并发安全的。与它协作的API都可以由外部控制执行【取消】操作,比如取消一个HTTP请求的执行。

Context

context.Background

  • 获取上下文根节点

Goroutine的创建和调用关系总是层层递进的,更靠顶部的Goroutine应该能够主动关闭其下属的子Goroutine,不然程序就可能会失控。为了实现这种关系,Context的结构也应该类似一棵树,叶子节点总数由根节点衍生出来的。

要创建Context树,首先需要获得根节点,context.Background()函数的返回值即根节点。

var (
    background = new(emptyCtx)
)

func Background() Context {
    return background
}

context.Background()函数会返回空的Context,该Context由接收请求的第一个Goroutine创建,是与进入请求对应的Context根节点,因此它不能被取消,同时也没有值和过期时间。常常作为处理请求的顶层context而存在。

context.TODO

  • 创建一个空的上下文
func TODO() Context {
    return todo
}

当不确定使用哪一种Context或准备接收一个Context时,可创建一个空的Context。

创建子节点

context包提供了4个函数来创建子节点、孙节点

func WithCancel(parent Context) (ctx Context, cancel CancelFunc) 
func WithDeadline(parent Context, d time.Time) (Context, CancelFunc)
func WithTimeout(parent Context, timeout time.Duration) (Context, CancelFunc) 
func WithValue(parent Context, key, val interface{}) Context 

创建函数都会接收一个Context类的参数parent作为父节点同时会返回一个Context的值作为子节点,以此层层向下创建出不同类型的子节点。

实际上子节点是从父节点复制而来的,根绝接收参数来设置子节点的不同的状态值,然后再将子节点传递给下层的Goroutine。

context.WithCancel

使用Context的Goroutine是无法取消某个操作的,因为Goroutine是被某个父Goroutine创建,因此只有父Goroutine才可以取消操作。不过,在父Goroutine中可通过WithCancel方法获得一个cancel方法,以此获得cancel子节点的权利。

func WithCancel(parent Context) (ctx Context, cancel CancelFunc) {
    if parent == nil {
        panic("cannot create context from nil parent")
    }
    c := newCancelCtx(parent)
    propagateCancel(parent, &c)
    return &c, func() { c.cancel(true, Canceled) }
}

WithCancel函数会将父节点复制给子节点后生成一个额外的CancelFunc函数类型的变量。

type CancelFunc func()

调用CancelFunc将主动撤销对应的Context对象

在父节点的Context对应的环境中可通过WithCancel函数不经可以创建子节点的Context,同时会获得该节点Context的控制权。一旦执行cancel函数,该节点Context就会结束。因此子节点需要判断是否结束来退出对应的Goroutine。

select {
    case <-cxt.Done():
        // 清理工作
}

例如:Goroutine泄露

阻塞两个子Goroutine,预期只有一个Goroutine能正常退出。

//LeakRoutine
func LeakRoutine() int{
    ch := make(chan int)
    //启动3个goroutine去抢占输入到通道
    go func(){
        ch<-1
    }()
    go func(){
        ch<-2
    }()
    go func(){
        ch<-3
    }()
    //一旦有输入立即返回
    return <-ch
}

func main(){
    //每层循环泄露两个goroutine
    for i:=0; i<4; i++{
        LeakRoutine()
        fmt.Printf("goroutines in loop end: %d\n", runtime.NumGoroutine())
    }
}

随着循环次数增加,除去mainGoroutine,每一轮循环都会泄露2个子Goroutine。

引入Context来管理子Goroutine

//FixLeakingByContext 使用上下文管理子协程避免协程泄露
func FixLeakingByContext() {
    //创建上下文用来管理子协程
    root := context.Background()//获取根节点
    ctx, cancel := context.WithCancel(root)//创建可cancel的子节点
    //函数结束前清理未结束的子协程
    defer cancel()

    //多协程强张
    ch := make(chan int)
    go CancelByContext(ctx, ch)
    go CancelByContext(ctx, ch)
    go CancelByContext(ctx, ch)

    //随机触发某个子协程退出
    ch <- 1
}
//CancelByContext 对管理进行轮询
func CancelByContext(ctx context.Context, ch chan (int)) int {
    select {
    //上下文通知结束
    case <-ctx.Done():
        return 0
    //接收到管道传入的值
    case n := <-ch:
        return n
    }
}

func main() {
    for i := 0; i < 4; i++ {
        //使用上下文管理子协程避免协程泄露
        FixLeakingByContext()
        //异步清理协程需时间
        time.Sleep(100)
        //每轮循环后只剩下主协程
        fmt.Printf("goroutines is loop end:%d\n", runtime.NumGoroutine())
    }
}

上游任务仅仅使用context通知下游任务不再需要但不会直接干涉会中断下游任务,由于下游任务自行决定后续的处理操作,也就是说context的取消操作是无侵入的。

例如:RPC调用

Main Goroutine上有4个RPC1/2/3/4,其中RPC2/3/4是并行请求的,这里希望在RPC2请求失败后字节返回错误,同时让RPC3/4停止继续计算。

RPC调用关系
  • 使用sync.WaitGroup保证main函数在所有RPC调用之后才调用
package main

import (
    "context"
    "errors"
    "fmt"
    "sync"
)

func RPC(ctx context.Context, url string) error {
    result := make(chan int)
    err := make(chan error)

    //执行远程过程调用
    go func() {
        ok := false
        if ok {
            result <- 1
        } else {
            err <- errors.New("rpc error")
        }
    }()

    select {
    //其它RPC调用失败
    case <-ctx.Done():
        return ctx.Err()
    //本RPC调用失败
    case e := <-err:
        return e
    //本RPC调用成功
    case <-result:
        return nil
    }
}

func main() {

    //创建顶级Context节点
    root := context.Background()
    //通过顶级创建带取消的取消权力的子节点
    ctx, cancel := context.WithCancel(root)

    //创建RPC1
    err := RPC(ctx, "http://rpc_1_url")
    fmt.Printf("RPC1:%v\n", err)
    if err != nil {
        return
    }

    //创建并行请求的RPC2/RPC3/RPC4
    wg := sync.WaitGroup{}
    wg.Add(1)
    go func() {
        defer wg.Done()
        err := RPC(ctx, "http://rpc_2_url")
        fmt.Printf("RPC2:%v\n", err)
        if err != nil {
            cancel() //撤销
        }
    }()
    wg.Add(1)
    go func() {
        defer wg.Done()
        err := RPC(ctx, "http://rpc_3_url")
        fmt.Printf("RPC3:%v\n", err)
        if err != nil {
            cancel() //撤销
        }
    }()
    wg.Add(1)
    go func() {
        defer wg.Done()
        err := RPC(ctx, "http://rpc_4_url")
        fmt.Printf("RPC3:%v\n", err)
        if err != nil {
            cancel() //撤销
        }
    }()
    wg.Wait()
}

RPC函数中,第一个参数是一个CancelContext,它相当于一个传话筒在创建时会返回一个听声器(ctx)和话筒(cancel)。

若有的Goroutine都要拿着听声器(ctx),当主Goroutine想要告诉所有Goroutine要结束时,通过cancel函数把结束信息告知所有Goroutine。

所有的Goroutine需内置处理听声器结束信号的逻辑ctx->Done(),在RPC函数内部通过一个select来判断ctxdone和当前RPC调用哪个先结束。

WaitGroup和其中一个RPC调用通知所有RPC逻辑,可直接采用errorGroup

contex.WithTimeout

  • context.WithTimeout用于控制子Goroutine的执行时间,可创建具有超时通知机制的Context对象。
func WithTimeout(parent Context, timeout time.Duration) (Context, CancelFunc) {
    return WithDeadline(parent, time.Now().Add(timeout))
}

例如:

parent := context.Background()
timeout := time.Millisecond * 3
ctx,cancel := context.WithTimeout(parent, timeout)
defer cancel()

select{
case <-time.After(time.Second * 1):
    fmt.Printf("OverSlept\n")
case <-ctx.Done():
    err := ctx.Err()
    fmt.Println(err)//context deadline exceeded
}

例如:HTTP客户端连接时添加超时处理

uri := "https://httpbin.org/delay/3"
req,err := http.NewRequest("GET", uri, nil)
if err!=nil {
    log.Fatalf("HTTP Request ERROR:%s\n", err)
}

parent := context.Background()
timeout := time.Millisecond * 100
ctx,_ := context.WithTimeout(parent, timeout)
req = req.WithContext(ctx)

resp,err := http.DefaultClient.Do(req)
if err!=nil {
    log.Fatalf("CLIENT DO ERROR:%s\n", err)
}
defer resp.Body.Close()

例如:HTTP服务端设置超时处理

  • http.TimeoutHandler本质上也是通过context.WithTimeout来做处理的
func defaultHandler(w http.ResponseWriter, r *http.Request){
    time.Sleep(time.Second * 10)
    w.Write([]byte("hello world"))
}

func main() {
    http.HandleFunc("/", defaultHandler)

    timeoutHandler := http.TimeoutHandler(http.DefaultServeMux, time.Second * 5, "timeout")

    http.ListenAndServe(":8888", timeoutHandler)
}

例如:超时请求

发送RPC请求时往往希望对请求进行超时限制,当一个RPC请求超过10s自动断开。

使用CancelContext可实现,需开启一个新的Goroutine获取cancel函数,当时间到了就调用cancel。

context.WithDeadline

context.WithValue


有疑问加站长微信联系(非本文作者)

本文来自:简书

感谢作者:JunChow520

查看原文:Go context

入群交流(和以上内容无关):加入Go大咖交流群,或添加微信:liuxiaoyan-s 备注:入群;或加QQ群:692541889

446 次点击  
加入收藏 微博
暂无回复
添加一条新回复 (您需要 登录 后才能回复 没有账号 ?)
  • 请尽量让自己的回复能够对别人有帮助
  • 支持 Markdown 格式, **粗体**、~~删除线~~、`单行代码`
  • 支持 @ 本站用户;支持表情(输入 : 提示),见 Emoji cheat sheet
  • 图片支持拖拽、截图粘贴等方式上传