context.WithCancel()的使用

留言

WithCancel可以将一个Context包装为cancelCtx,并提供一个取消函数,调用这个取消函数,可以Cancel对应的Context

Go语言context包-cancelCtx


疑问


context.WithCancel()取消机制的理解

父母5s钟后出门，倒计时，父母在时要学习，父母一走就可以玩

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
package main

import (
	"context"
	"fmt"
	"time"
)

func dosomething(ctx context.Context) {
	for {
		select {
		case <-ctx.Done():
			fmt.Println("playing")
			return
		default:
			fmt.Println("I am working!")
			time.Sleep(time.Second)
		}
	}
}

func main() {
	ctx, cancelFunc := context.WithCancel(context.Background())
	go func() {
		time.Sleep(5 * time.Second)
		cancelFunc()
	}()
	dosomething(ctx)
}

为什么调用cancelFunc就能从ctx.Done()里取得返回值? 进而取消对应的Context？


复习一下channel的一个特性


从一个已经关闭的channel里可以一直获取对应的零值


WithCancel代码分析


pkg.go.dev/context#WithCancel:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
// WithCancel returns a copy of parent with a new Done channel. The returned
// context's Done channel is closed when the returned cancel function is called
// or when the parent context's Done channel is closed, whichever happens first.
//
// Canceling this context releases resources associated with it, so code should
// call cancel as soon as the operations running in this Context complete.

//WithCancel 返回具有新 Done 通道的 parent 副本。 返回的上下文的完成通道在调用返回的取消函数或父上下文的完成通道关闭时关闭，以先发生者为准。

//取消此上下文会释放与其关联的资源，因此代码应在此上下文中运行的操作完成后立即调用取消。

func WithCancel(parent Context) (ctx Context, cancel CancelFunc) {
	if parent == nil {
		panic("cannot create context from nil parent")
	}
	c := newCancelCtx(parent)   // 将parent作为父节点context 生成一个新的子节点

	//获得“父Ctx路径”中可被取消的Ctx
	//将child canceler加入该父Ctx的map中
	propagateCancel(parent, &c)
	return &c, func() { c.cancel(true, Canceled) }
}

WithCancel最后返回 子上下文和一个cancelFunc函数，而cancelFunc函数里调用了cancelCtx这个结构体的方法cancel

(代码基于go 1.16； 1.17有所改动)

1
2
3
4
5
6
7
8
9
10
// A cancelCtx can be canceled. When canceled, it also cancels any children
// that implement canceler.
type cancelCtx struct {
	Context

	mu       sync.Mutex            // protects following fields
	done     chan struct{}         // created lazily, closed by first cancel call done是一个channel，用来 传递关闭信号
	children map[canceler]struct{} // set to nil by the first cancel call  children是一个map，存储了当前context节点下的子节点
	err      error                 // set to non-nil by the first cancel call  err用于存储错误信息 表示任务结束的原因
}

在cancelCtx这个结构体中，字段done是一个传递空结构体类型的channel，用来在上下文取消时关闭这个通道，err就是在上下文被取消时告诉用户这个上下文取消了，可以用ctx.Err()来获取信息

canceler是一个实现接口，用于Ctx的终止。实现该接口的Context有cancelCtx和timerCtx，而emptyCtx和valueCtx没有实现该接口。

1
2
3
4
5
6
// A canceler is a context type that can be canceled directly. The
// implementations are *cancelCtx and *timerCtx.
type canceler interface {
	cancel(removeFromParent bool, err error)
	Done() <-chan struct{}
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46

// closedchan is a reusable closed channel.
var closedchan = make(chan struct{})

func init() {
	close(closedchan)
}

// cancel closes c.done, cancels each of c's children, and, if
// removeFromParent is true, removes c from its parent's children.
/**
* 1、cancel(...)当前Ctx的子节点
* 2、从父节点中移除该Ctx
**/
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

	//  设置一个关闭的channel或者将done channel关闭，用以发送关闭信号
	if c.done == nil {
		c.done = closedchan
	} else {
		close(c.done) // 注意这一步
	}

	 // 将子节点context依次取消
	for child := range c.children {
		// NOTE: acquiring the child's lock while holding parent's lock.
		child.cancel(false, err)
	}
	c.children = nil
	c.mu.Unlock()

	if removeFromParent {
		 // 将当前context节点从父节点上移除
		removeChild(c.Context, c)
	}
}

对于cancel函数，其取消了基于该上下文的所有子上下文以及把自身从父上下文中取消

对于更多removeFromParent代码分析,和其他Context的使用，强烈建议阅读 深入理解Golang之Context（可用于实现超时机制）


1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
// Done is provided for use in select statements:
//
//  // Stream generates values with DoSomething and sends them to out
//  // until DoSomething returns an error or ctx.Done is closed.
//  func Stream(ctx context.Context, out chan<- Value) error {
//  	for {
//  		v, err := DoSomething(ctx)
//  		if err != nil {
//  			return err
//  		}
//  		select {
//  		case <-ctx.Done():
//  			return ctx.Err()
//  		case out <- v:
//  		}
//  	}
//  }
//
// See https://blog.golang.org/pipelines for more examples of how to use
// a Done channel for cancellation.
Done() <-chan struct{}

// If Done is not yet closed, Err returns nil.
// If Done is closed, Err returns a non-nil error explaining why:
// Canceled if the context was canceled
// or DeadlineExceeded if the context's deadline passed.
// After Err returns a non-nil error, successive calls to Err return the same error.
Err() error

当调用cancelFunc()时，会有一步close(d)的操作,

ctx.Done 获取一个只读的 channel，类型为结构体。可用于监听当前 channel 是否已经被关闭。

Done()用来监听cancel操作(对于cancelCtx)或超时操作(对于timerCtx)，当执行取消操作或超时时，c.done会被close，这样就能从一个已经关闭的channel里一直获取对应的零值，<-ctx.Done便不会再阻塞

(代码基于go 1.16； 1.17有所改动)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
func (c *cancelCtx) Done() <-chan struct{} {
	c.mu.Lock()
	if c.done == nil {
		c.done = make(chan struct{})
	}
	d := c.done
	c.mu.Unlock()
	return d
}

func (c *cancelCtx) Err() error {
	c.mu.Lock()
	err := c.err
	c.mu.Unlock()
	return err
}

总结一下：使用context.WithCancel时，除了返回一个新的context.Context(上下文)，还会返回一个cancelFunc。 在需要取消该context.Context时，就调用这个cancelFunc,之后当前上下文及其子上下文都会被取消，所有的 Goroutine 都会同步收到这一取消信号

至于cancelFunc是如何做到的？

在用户代码，for循环里select不断尝试从 <-ctx.Done()里读取出内容，但此时并没有任何给 c.done这个channel写入数据的操作，（类似c.done <- struct{}{}）,故而在for循环里每次select时，这个case都不满足条件，一直阻塞着。每次都执行default代码段

而在执行cancelFunc时， 在func (c *cancelCtx) cancel(removeFromParent bool, err error)里面，会有一个close(c.done)的操作。而从一个已经关闭的channel里可以一直获取对应的零值，即 select可以命中，进入case res := <-ctx.Done():代码段


可用如下代码验证：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
package main

import (
	"context"
	"fmt"
	"time"
)

func dosomething(ctx context.Context) {

	var cuiChan = make(chan struct{})

	go func() {
		cuiChan <- struct{}{}
	}()

	//close(cuiChan)

	for {
		select {
		case res := <-ctx.Done():
			fmt.Println("res:", res)
			return
		case res2 := <-cuiChan:
			fmt.Println("res2:", res2)
		default:
			fmt.Println("I am working!")
			time.Sleep(time.Second)
		}
	}
}

func main() {

	test()
	ctx, cancelFunc := context.WithCancel(context.Background())
	go func() {
		time.Sleep(5 * time.Second)
		cancelFunc()
	}()

	dosomething(ctx)
}

func test() {

	var testChan = make(chan struct{})

	if testChan == nil {
		fmt.Println("make(chan struct{})后为nil")
	} else {
		fmt.Println("make(chan struct{})后不为nil！！！")
	}

}

输出：

1
2
3
4
5
6
7
8
make(chan struct{})后不为nil！！！
I am working!
res2: {}
I am working!
I am working!
I am working!
I am working!
res: {}

而如果 不向没有缓存的cuiChan写入数据，直接close，即

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
package main

import (
	"context"
	"fmt"
	"time"
)

func dosomething(ctx context.Context) {

	var cuiChan = make(chan struct{})

	//go func() {
	//	cuiChan <- struct{}{}
	//}()

	close(cuiChan)

	for {
		select {
		case res := <-ctx.Done():
			fmt.Println("res:", res)
			return
		case res2 := <-cuiChan:
			fmt.Println("res2:", res2)
		default:
			fmt.Println("I am working!")
			time.Sleep(time.Second)
		}
	}
}

func main() {

	test()
	ctx, cancelFunc := context.WithCancel(context.Background())
	go func() {
		time.Sleep(5 * time.Second)
		cancelFunc()
	}()

	dosomething(ctx)
}

func test() {

	var testChan = make(chan struct{})

	if testChan == nil {
		fmt.Println("make(chan struct{})后为nil")
	} else {
		fmt.Println("make(chan struct{})后不为nil！！！")
	}

}

则会一直命中case 2

1
2
3
4
5
6
7
8
9
res2: {}
res2: {}
res2: {}
res2: {}
res2: {}
res2: {}
res2: {}
...
//一直打印下去

更多参考：

深入理解Golang之Context（可用于实现超时机制）

回答我，停止 Goroutine 有几种方法？

golang context的done和cancel的理解 for循环channel实现context.Done()阻塞输出



更多关于channel阻塞与close的代码


1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
package main

import (
	"fmt"
	"time"
)

func main() {
	ch := make(chan string, 0)
	go func() {
		for {
			fmt.Println("----开始----")
			v, ok := <-ch
			fmt.Println("v,ok", v, ok)
			if !ok {
				fmt.Println("结束")
				return
			}
			//fmt.Println(v)
		}
	}()

	fmt.Println("<-ch一直没有东西写进去，会一直阻塞着,直到3秒钟后")
	fmt.Println()
	fmt.Println()
	time.Sleep(3 * time.Second)

	ch <- "向ch这个channel写入第一条数据..."
	ch <- "向ch这个channel写入第二条数据！！！"

	close(ch) // 当channel被close后， v,ok 中的ok就会变为false

	time.Sleep(10 * time.Second)
}

输出为：

1
2
3
4
5
6
7
8
9
10
----开始----
<-ch一直没有东西写进去，会一直阻塞着,直到3秒钟后


v,ok 向ch这个channel写入第一条数据... true
----开始----
v,ok 向ch这个channel写入第二条数据！！！ true
----开始----
v,ok  false
结束


1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
package main

import (
	"fmt"
	"sync/atomic"
	"time"
)

func main() {
	ch := make(chan string, 0)
	done := make(chan struct{})

	go func() {
		var i int32

		for {
			atomic.AddInt32(&i, 1)
			select {
			case ch <- fmt.Sprintf("%s%d%s", "第", i, "次向通道中写入数据"):

			case <-done:
				close(ch)
				return
			}

			// select随机选择满足条件的case，并不按顺序，所以打印出的结果，在30几次波动
			time.Sleep(100 * time.Millisecond)
		}
	}()

	go func() {
		time.Sleep(3 * time.Second)
		done <- struct{}{}
	}()

	for i := range ch {
		fmt.Println("接收到的值: ", i)
	}

	fmt.Println("结束")
}

输出为：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
接收到的值:  第1次向通道中写入数据
接收到的值:  第2次向通道中写入数据
接收到的值:  第3次向通道中写入数据
接收到的值:  第4次向通道中写入数据
接收到的值:  第5次向通道中写入数据
接收到的值:  第6次向通道中写入数据
接收到的值:  第7次向通道中写入数据
接收到的值:  第8次向通道中写入数据
接收到的值:  第9次向通道中写入数据
接收到的值:  第10次向通道中写入数据
接收到的值:  第11次向通道中写入数据
接收到的值:  第12次向通道中写入数据
接收到的值:  第13次向通道中写入数据
接收到的值:  第14次向通道中写入数据
接收到的值:  第15次向通道中写入数据
接收到的值:  第16次向通道中写入数据
接收到的值:  第17次向通道中写入数据
接收到的值:  第18次向通道中写入数据
接收到的值:  第19次向通道中写入数据
接收到的值:  第20次向通道中写入数据
接收到的值:  第21次向通道中写入数据
接收到的值:  第22次向通道中写入数据
接收到的值:  第23次向通道中写入数据
接收到的值:  第24次向通道中写入数据
接收到的值:  第25次向通道中写入数据
接收到的值:  第26次向通道中写入数据
接收到的值:  第27次向通道中写入数据
接收到的值:  第28次向通道中写入数据
接收到的值:  第29次向通道中写入数据
接收到的值:  第30次向通道中写入数据
接收到的值:  第31次向通道中写入数据
结束

每次执行，打印出的结果，在30几次波动