Golang 并发 与 context标准库

这篇文章将:介绍context工作机制;简单说明接口和结构体功能;通过简单Demo介绍外部API创建并使用context标准库;从源码角度分析context工作流程(不包括mutex的使用分析以及timerCtx计时源码)。

context是一个很好的解决多goroutine下通知传递和元数据的Go标准库。由于Go中的goroutine之间没有父子关系,因此也不存在子进程退出后的通知机制。多个goroutine协调工作涉及 通信同步通知退出 四个方面:

通信:chan通道是各goroutine之间通信的基础。注意这里的通信主要指程序的数据通道。
同步:可以使用不带缓冲的chan;sync.WaitGroup为多个gorouting提供同步等待机制;mutex锁与读写锁机制。
通知:通知与上文通信的区别是,通知的作用为管理,控制流数据。一般的解决方法是在输入端绑定两个chan,通过select收敛处理。这个方案可以解决简单的问题,但不是一个通用的解决方案。
退出:简单的解决方案与通知类似,即增加一个单独的通道,借助chan和select的广播机制(close chan to broadcast)实现退出。

但由于Go之间的goroutine都是平等的,因此当遇到复杂的并发结构时处理退出机制则会显得力不从心。因此Go1.7版本开始提供了context标准库来解决这个问题。他提供两个功能:退出通知元数据传递。他们都可以传递给整个goroutine调用树的每一个goroutine。同时这也是一个不太复杂的,适合初学Gopher学习的一段源码。


工作机制

第一个创建Context的goroutine被称为root节点:root节点负责创建一个实现Context接口的具体对象,并将该对象作为参数传递至新拉起的goroutine作为其上下文。下游goroutine继续封装该对象并以此类推向下传递。

interface

Context接口:作为一个基本接口,所有的Context对象都要实现该接口,并将其作为使用者调度时的参数类型:

type Context interface{
    Deadline()(deadline time.Time, ok bool)  
//如果Context实现了超时控制,该方法返回 超时时间,true。否则ok为false
    Done() <-chan struct{}
//依旧使用<-chan struct{}来通知退出,供被调用的goroutine监听。
    Err() error
//当Done()返回的chan收到通知后,防卫Err()获知被取消的原因
    Value(key interface{}) interface
}

canceler接口:拓展接口,规定了取消通知的Context具体类型需要实现的接口:

type canceler interface {
    cancel(removeFromParent bool, err error)
//通知后续创建的goroutine退出
    Done() <-chan struct{}
//作者对这个Done()方法的理解是多余的
}

struct

emptyCtx:实现了一个不具备任何功能的Context接口,其存在的目的就是作为Context对象树的root节点:

type emptyCtx int
func (*emptyCtx) Deadline() (deadline time.Time, ok bool) {
    return
}
func (*emptyCtx) Done() <-chan struct{} {
    return nil
}
func (*emptyCtx) Err() error {
    return nil
}
func (*emptyCtx) Value(key interface{}) interface{} {
    return nil
}
//......
var (
    background = new(emptyCtx)
    todo       = new(emptyCtx)
)
func Background() Context {
    return background
}
func TODO() Context {
    return todo
}
//这两者返回值是一样的,文档上建议main函数可以使用Background()创建root context

cancelCtx:可以认为他与emptyCtx最大的区别在于,具体实现了cancel函数。即他可以向子goroutine传递cancel消息。
timerCtx:另一个实现Context接口的具体类型,内部封装了cancelCtx类型实例,同时拥有deadline变量,用于实现定时退出通知。
valueCtx:实现了Context接口的具体类型,内部分装cancelCtx类型实例,同时封装了一个kv存储变量,用于传递通知消息。

API

除了root context可以使用Background()创建以外,其余的context都应该从cancelCtxtimerCtxvalueCtx中选取一个来构建具体对象:
func WithCancel(parent Context) (Context, CancelFunc):创建cancelCtx实例。
func WithDeadline(parent Context, deadline time.Time)(Context, CancelFunc)func WithTimeout(parent Context, timeout time.Duration)(Context, CancelFunc):两种方法都可以创建一个带有超时通知的Context具体对象timerCtx,具体差别在于传递绝对或相对时间。
func WithValue(parent Context, key, val interface{}) Context:创建valueCtx实例。


  1. 创建root context并构建一个WithCancel类型的上下文,使用该上下文注册一个goroutine模拟运行:
func main(){
    ctxa, cancel := context.WithCancel(context.Background())
    go work(ctxa, "work1")
}
func work(ctx context.Context, name string){
    for{
        select{
        case <-ctx.Done():
            println(name," get message to quit")
            return
        default:
            println(name," is running")
            time.Sleep(time.Second)
        }
    }
}
  1. 使用WithDeadline包装ctxa,并使用新的上下文注册另一个goroutine:
func main(){
    ctxb, _ := context.WithTimeout(ctxa, time.Second * 3)
    go work(ctxb, "work2")
}
  1. 使用WithValue包装ctxb,并注册新的goroutine:
func main(){
    ctxc := context.WithValue(ctxb, "key", "custom value")
    go workWithValue(ctxc, "work3")
}
func workWithValue(ctx context.Context, name string){
    for{
        select {
        case <-ctx.Done():
            println(name," get message to quit")
            return
        default:
            value:=ctx.Value("key").(string)
            println(name, " is running with value", value)
            time.Sleep(time.Second)
        }
    }
}
  1. 最后在main函数中手动关闭ctxa,并等待输出结果:
func main(){
    time.Sleep(5*time.Second)
    cancel()
    time.Sleep(time.Second)
}

至此我们运行程序并查看输出结果:

work1  is running
work3  is running with value custom value
work2  is running
work1  is running
work2  is running
work3  is running with value custom value
work2  is running
work3  is running with value custom value
work1  is running
//work2超时并通知work3退出
work2  get message to quit
work3  get message to quit
work1  is running
work1  is running
work1  get message to quit

可以看到,当ctxb因超时而退出之后,会通知由他包装的所有子goroutine(ctxc),并通知退出。各context的关系结构如下:

Background() -> ctxa -> ctxb -> ctxc


源码分析

我们主要研究两个问题,即各Context如何保存父类和子类上下文;以及cancel方法如何实现通知子类context实现退出功能。

context的数据结构
  1. emptyCtx只是一个uint类型的变量,其目的只是为了作为第一个goroutine ctx的parent,因此他不需要,也没法保存子类上下文结构。

  2. cancelCtx的数据结构:

type cancelCtx struct {
    Context

    mu       sync.Mutex            // protects following fields
    done     chan struct{}         // created lazily, closed by first cancel call
    children map[canceler]struct{} // set to nil by the first cancel call
    err      error                 // set to non-nil by the first cancel call
}

Context接口保存的就是父类的context。children map[canceler]struct{}保存的是所有直属与这个context的子类context。done chan struct{}用于发送退出信号。
我们查看创建cancelCtx的APIfunc WithCancel(...)...

func WithCancel(parent Context) (ctx Context, cancel CancelFunc) {
    c := newCancelCtx(parent)
    propagateCancel(parent, &c)
    return &c, func() { c.cancel(true, Canceled) }
}
func newCancelCtx(parent Context) cancelCtx {
    return cancelCtx{Context: parent}
}

propagateCancel函数的作用是将自己注册至parent context。我们稍后会讲解这个函数。

  1. timerCtx的数据结构:
type timerCtx struct {
    cancelCtx
    timer *time.Timer // Under cancelCtx.mu.

    deadline time.Time
}

timerCtx继承于cancelCtx,并为定时退出功能新增自己的数据结构。

func WithDeadline(parent Context, d time.Time) (Context, CancelFunc) {
    if cur, ok := parent.Deadline(); ok && cur.Before(d) {
        // The current deadline is already sooner than the new one.
        return WithCancel(parent)
    }
    c := &timerCtx{
        cancelCtx: newCancelCtx(parent),
        deadline:  d,
    }
    propagateCancel(parent, c)
//以下内容与定时退出机制有关,在本文不作过多分析和解释
    dur := time.Until(d)
    if dur <= 0 {
        c.cancel(true, DeadlineExceeded) // deadline has already passed
        return c, func() { c.cancel(true, Canceled) }
    }
    c.mu.Lock()
    defer c.mu.Unlock()
    if c.err == nil {
        c.timer = time.AfterFunc(dur, func() {
            c.cancel(true, DeadlineExceeded)
        })
    }
    return c, func() { c.cancel(true, Canceled) }
}
func newCancelCtx(parent Context) cancelCtx {
    return cancelCtx{Context: parent}
}

timerCtx查看parent context的方法是timerCtx.cancelCtx.Context

  1. valueCtx的数据结构:
type valueCtx struct {
    Context
    key, val interface{}
}

相较于timerCtx而言非常简单,没有继承于cancelCtx struct,而是直接继承于Context接口。

func WithValue(parent Context, key, val interface{}) Context {
    if key == nil {
        panic("nil key")
    }
    if !reflect.TypeOf(key).Comparable() {
        panic("key is not comparable")
    }
    return &valueCtx{parent, key, val}
}

辅助函数

这里我们会有两个疑问,第一,valueCtx为什么没有propagateCancel函数向parent context注册自己。既然没有注册,为何ctxb超时后能通知ctxc一起退出。第二,valueCtx是如何存储children和parent context结构的。相较于同样绑定Context接口的cancelCtx,valueCtx并没有children数据。
第二个问题能解决一半第一个问题,即为何不向parent context注册。先说结论:valueCtx的children context注册在valueCtx的parent context上。函数func propagateCancel(...)负责注册信息,我们先看一下他的构造:

func propagateCancel

func propagateCancel(parent Context, child canceler) {
    if parent.Done() == nil {
        return // parent is never canceled
    }
    if p, ok := parentCancelCtx(parent); ok {
        p.mu.Lock()
        if p.err != nil {
            // parent has already been canceled
            child.cancel(false, p.err)
        } else {
            if p.children == nil {
                p.children = make(map[canceler]struct{})
            }
            p.children[child] = struct{}{}
        }
        p.mu.Unlock()
    } else {
        go func() {
            select {
            case <-parent.Done():
                child.cancel(false, parent.Err())
            case <-child.Done():
            }
        }()
    }
}

这个函数的主要逻辑如下:接收parent context 和 child canceler方法,若parent为emptyCtx,则不注册;否则通过funcparentCancelCtx寻找最近的一个*cancelCtx;若该cancelCtx已经结束,则调用child的cancel方法,否则向该cancelCtx注册child。

func parentCancelCtx

func parentCancelCtx(parent Context) (*cancelCtx, bool) {
    for {
        switch c := parent.(type) {
        case *cancelCtx:
            return c, true
        case *timerCtx:
            return &c.cancelCtx, true
        case *valueCtx:
            parent = c.Context
        default:
            return nil, false
        }
    }
}

func parentCancelCtx从parentCtx中向上迭代寻找第一个*cancelCtx并返回。从函数逻辑中可以看到,只有当parent.(type)为*valueCtx的时候,parent才会向上迭代而不是立即返回。否则该函数都是直接返回或返回经过包装的*cancelCtx。因此我们可以发现,valueCtx是依赖于parentCtx的*cancelCtx结构的。

至于第二个问题,事实上,parentCtx根本无需,也没有办法通过Done()方法通知valueCtx,valueCtx也没有额外实现Done()方法。可以理解为:valueCtx与parentCtx公用一个done channel,当parentCtx调用了cancel方法并关闭了done channel时,监听valueCtx的done channel的goroutine同样会收到退出信号。另外,当parentCtx没有实现cancel方法(如emptyCtx)时,可以认为valueCtx也是无法cancel的。

func (c *cancelCtx) cancel

func (c *cancelCtx) cancel(removeFromParent bool, err error) {
    if err == nil {
        panic("context: internal error: missing cancel error")
    }
    c.mu.Lock()
    if c.err != nil {
        c.mu.Unlock()
        return // already canceled
    }
    c.err = err
    if c.done == nil {
        c.done = closedchan
    } else {
        close(c.done)
    }
    for child := range c.children {
        child.cancel(false, err)
    }
    c.children = nil
    c.mu.Unlock()

    if removeFromParent {
        removeChild(c.Context, c)
    }
}

该方法的主要逻辑如下:若外部err为空,则代表这是一个非法的cancel操作,抛出panic;若cancelCtx内部err不为空,说明该Ctx已经执行过cancel操作,直接返回;关闭done channel,关联该Ctx的goroutine收到退出通知;遍历children,若有的话,执行child.cancel操作;调用removeChild将自己从parent context中移除。

func (c *timerCtx) cancel

与cancelCtx十分类似,不作过多阐述。

最后编辑于
©著作权归作者所有,转载或内容合作请联系作者
  • 序言:七十年代末,一起剥皮案震惊了整个滨河市,随后出现的几起案子,更是在滨河造成了极大的恐慌,老刑警刘岩,带你破解...
    沈念sama阅读 203,324评论 5 476
  • 序言:滨河连续发生了三起死亡事件,死亡现场离奇诡异,居然都是意外死亡,警方通过查阅死者的电脑和手机,发现死者居然都...
    沈念sama阅读 85,303评论 2 381
  • 文/潘晓璐 我一进店门,熙熙楼的掌柜王于贵愁眉苦脸地迎上来,“玉大人,你说我怎么就摊上这事。” “怎么了?”我有些...
    开封第一讲书人阅读 150,192评论 0 337
  • 文/不坏的土叔 我叫张陵,是天一观的道长。 经常有香客问我,道长,这世上最难降的妖魔是什么? 我笑而不...
    开封第一讲书人阅读 54,555评论 1 273
  • 正文 为了忘掉前任,我火速办了婚礼,结果婚礼上,老公的妹妹穿的比我还像新娘。我一直安慰自己,他们只是感情好,可当我...
    茶点故事阅读 63,569评论 5 365
  • 文/花漫 我一把揭开白布。 她就那样静静地躺着,像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上,一...
    开封第一讲书人阅读 48,566评论 1 281
  • 那天,我揣着相机与录音,去河边找鬼。 笑死,一个胖子当着我的面吹牛,可吹牛的内容都是我干的。 我是一名探鬼主播,决...
    沈念sama阅读 37,927评论 3 395
  • 文/苍兰香墨 我猛地睁开眼,长吁一口气:“原来是场噩梦啊……” “哼!你这毒妇竟也来了?” 一声冷哼从身侧响起,我...
    开封第一讲书人阅读 36,583评论 0 257
  • 序言:老挝万荣一对情侣失踪,失踪者是张志新(化名)和其女友刘颖,没想到半个月后,有当地人在树林里发现了一具尸体,经...
    沈念sama阅读 40,827评论 1 297
  • 正文 独居荒郊野岭守林人离奇死亡,尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 35,590评论 2 320
  • 正文 我和宋清朗相恋三年,在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 37,669评论 1 329
  • 序言:一个原本活蹦乱跳的男人离奇死亡,死状恐怖,灵堂内的尸体忽然破棺而出,到底是诈尸还是另有隐情,我是刑警宁泽,带...
    沈念sama阅读 33,365评论 4 318
  • 正文 年R本政府宣布,位于F岛的核电站,受9级特大地震影响,放射性物质发生泄漏。R本人自食恶果不足惜,却给世界环境...
    茶点故事阅读 38,941评论 3 307
  • 文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹,春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 29,928评论 0 19
  • 文/苍兰香墨 我抬头看了看天上的太阳。三九已至,却和暖如春,着一层夹袄步出监牢的瞬间,已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 31,159评论 1 259
  • 我被黑心中介骗来泰国打工, 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留,地道东北人。 一个月前我还...
    沈念sama阅读 42,880评论 2 349
  • 正文 我出身青楼,却偏偏与公主长得像,于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子,可洞房花烛夜当晚...
    茶点故事阅读 42,399评论 2 342

推荐阅读更多精彩内容