GoLong的学习之路，进阶，标准库之并发（context）补充并发三部曲，你真的明白context吗？

其实对于，context来说，如果只是用来做并发处理就有些不太合适。因为对于golang来说，context应用场景不仅在并发有用，并且在网络链接，http处理，gorm中都有体现。但是其实，本质来说。以上这些场景其实都是并发goroutine 的应用。故在这里我只讲context在并发的情况。

Context

Context这个其实在其他语言都有涉及，比如：

• Spirngboot ：ApplicationContext
• flutter： BuildContext
• Android：Context
• Kotlin： CoroutineContext
• .NET： CancellationSourceToken

可以说很多很多了。但是他们的用法不尽相同。而 go中的 context 的设计思想以及应用场景非常的新颖。为什么这么说呢？这就要说go中最大的特点：goroutine 协程。

go以协程而闻名。

而如何有效控制goroutine （协程），这个问题有很多不同解释：

比如：

• （1）有效控制：协程之间对于共享变量的控制------------：> 加锁操作
• （2）有效控制：协程之间数据的交互------------------------：> channle
• （3）有效控制：协程之间先后顺序---------------------------：> 阻塞（select，sycn包）
• （4）有效控制：协程之间错误信息---------------------------：> recover、errgroup
• （5）有效控制：协程之间的嵌套------------------------------：> context(今天要重点说明的)

上面的这些1，2，3，4点在我的并发三部曲中有体现。感兴趣的可以移步

Context的出现

一个场景问题：如何在主协程中，控制，从协程的开启和结束呢？

方案一、

用时间去控制（很明显这种方式，并不是很恰当）（时间设置过早或者过晚都有问题，很难控制）

方案二、

提供一个全局变量，用这个全局变量去控制从协程的开始和结束

package main

import (
	"fmt"
	"sync"
	"time"
)

var wg sync.WaitGroup
var exit bool

func worker() {
	for {
		fmt.Println("worker")
		time.Sleep(time.Second)
		if exit {
			break
		}
	}
	wg.Done()
}

func main() {
	wg.Add(1)
	go worker()
	time.Sleep(time.Second * 3) // sleep3秒以免程序过快退出
	exit = true                 // 修改全局变量实现子goroutine的退出
	wg.Wait()
	fmt.Println("over")
}
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

全局变量方式存在的问题：

1. 使用全局变量在跨包调用时不容易统一
2. 如果worker中再启动goroutine，就不太好控制了

方案三、

既然能通过全局变量的方式去达到控制的目的，那么是不是可以用channle

通过通道（channle）去控制一个局部变量，用接收和发送的方式达到控制协程的目的

package main

import (
	"fmt"
	"sync"

	"time"
)

var wg sync.WaitGroup



func worker(exitChan chan struct{}) {
LOOP:
	for {
		fmt.Println("worker")
		time.Sleep(time.Second)
		select {
		case <-exitChan: // 等待接收上级通知
			break LOOP
		default:
		}
	}
	wg.Done()
}

func main() {
	var exitChan = make(chan struct{})
	wg.Add(1)
	go worker(exitChan)
	time.Sleep(time.Second * 3) // sleep3秒以免程序过快退出
	exitChan <- struct{}{}      // 给子goroutine发送退出信号
	close(exitChan)
	wg.Wait()
	fmt.Println("over")
}
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

管道方式存在的问题：

1. 使用全局变量跨包调用时不容易实现规范和统一，需要维护一个共用的channel(太麻烦了)

---

综合：

就这些传统控制的手段，相信大家看见代码和我的解释，一定有所感受，但是我要说的是。这些解决方案，并非不可行。对于少量的goroutine来说这种方式未尝不可。

但是一旦嵌套了很多层。我们对于这种方式真的能合理控制—达到不 “ 然 ” 吗？（糊涂，混乱的意思）大佬除外

任务的 goroutine 层级越深，想要自己做退出信号感知和元数据共享就越难

---

所以我们需要一种优雅的方案来实现这样一种机制：

• 上层任务取消后，所有的下层任务都会被取消
• 中间某一层的任务取消后，只会将当前任务的下层任务取消，而不会影响上层的任务以及同级任务
• 可以线程安全地在 goroutine 之间共享一些任务的元数据

所以为此 Go 官方在1.7 版本引入了 Context 来实现上面阐述的机制

context的设计思想

我们先看源码接口：

type Context interface {

    Deadline() (deadline time.Time, ok bool)

    Done() <-chan struct{}

    Err() error

    Value(key interface{}) interface{}
}
1
2
3
4
5
6
7
8
9
10

Context接口包含四个方法：

• Deadline返回绑定当前context的任务被取消的截止时间；如果没有设定期限，将返回ok == false。
• Done 当绑定当前context的任务被取消时，将返回一个关闭的channel；如果当前context不会被取消，将返回nil。
• Err 如果Done返回的channel没有关闭，将返回nil;如果Done返回的channel已经关闭，将返回非空的值表示任务结束的原因。如果是context被取消，Err将返回Canceled；如果是context超时，Err将返回DeadlineExceeded。
• Value 返回context存储的键值对中当前key对应的值，如果没有对应的key,则返回nil。

---

1. Done方法返回的channel正是用来传递结束信号以抢占并中断当前任务；
2. Deadline方法指示一段时间后当前goroutine是否会被取消；
3. Err方法，来解释goroutine被取消的原因；
   • 如果当前Context被取消就会返回Canceled错误；
   • 如果当前Context超时就会返回DeadlineExceeded错误；
4. Value则用于获取特定于当前任务树的额外信息；

首先我要说的是context的结构是一个树状结构。为了方便找到根节点，有定义了一个结构emptyCtx

emptyCtx是一个int类型的变量，但实现了context的接口。emptyCtx没有超时时间，不能取消，也不能存储任何额外信息，所以emptyCtx用来作为context树的根节点。

// An emptyCtx is never canceled, has no values, and has no deadline. It is not
// struct{}, since vars of this type must have distinct addresses.
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
}

func (e *emptyCtx) String() string {
    switch e {
    case background:
        return "context.Background"
    case todo:
        return "context.TODO"
    }
    return "unknown empty Context"
}

var (
    background = new(emptyCtx)
    todo       = new(emptyCtx)
)
func Background() Context {
    return background
}
func TODO() Context {
    return todo
}
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

但我们一般不会直接使用emptyCtx，而是使用由emptyCtx实例化的两个变量，分别可以通过调用Background和TODO方法得到

每次要在Context链路上增加要携带的键值对时，都要在上级Context的基础上新建一个 valueCtx 存储键值对，切只能增加不能修改，读取 Context 上的键值又是一个幂等的操作，所以 Context 就这样实现了线程安全的数据共享机制，且全程无锁，不会影响性能。

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
}

type canceler interface {
    cancel(removeFromParent bool, err error)
    Done() <-chan struct{}
}
1
2
3
4
5
6
7
8
9
10
11
12
13

• cancelCtx中也有一个context变量作为父节点
• 变量done表示一个channel，用来表示传递关闭信号
• children表示一个map，存储了当前context节点下的子节点
• err用于存储错误信息表示任务结束的原因

valueCtx结构体

type valueCtx struct {
    Context
    key, val interface{}
}

func (c *valueCtx) Value(key interface{}) interface{} {
    if c.key == key {
        return c.val
    }
    return c.Context.Value(key)
}
1
2
3
4
5
6
7
8
9
10
11

• valueCtx利用一个Context类型的变量来表示父节点context，所以当前context继承了父context的所有信息
• valueCtx类型还携带一组键值对，也就是说这种context可以携带额外的信息
• valueCtx实现了Value方法，用以在context链路上获取key对应的值，如果当前context上不存在需要的key,会沿着context链向上寻找key对应的值，直到根节点

timerCtx结构体

type timerCtx struct {
    cancelCtx
    timer *time.Timer // Under cancelCtx.mu.

    deadline time.Time
}

func (c *timerCtx) Deadline() (deadline time.Time, ok bool) {
    return c.deadline, true
}

func (c *timerCtx) cancel(removeFromParent bool, err error) {
    将内部的cancelCtx取消
    c.cancelCtx.cancel(false, err)
    if removeFromParent {
        // Remove this timerCtx from its parent cancelCtx's children.
        removeChild(c.cancelCtx.Context, c)
    }
    c.mu.Lock()
    if c.timer != nil {
        取消计时器
        c.timer.Stop()
        c.timer = nil
    }
    c.mu.Unlock()
}
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

• timerCtx内部使用cancelCtx实现取消，另外使用定时器timer和过期时间deadline实现定时取消的功能。
• timerCtx在调用cancel方法，会先将内部的cancelCtx取消，如果需要则将自己从cancelCtx祖先节点上移除，最后取消计时器。

---

除了Context 接口外还定义了一个叫做 canceler 的接口，实现了它的类型即为带取消功能的 Context。

• emptyCtx 什么属性也没有，啥也不能干。
• valueCtx 只能携带一个键值对，且依附在上一级 Context 上。
• timerCtx 继承自 cancelCtx 他们都是带取消功能的 Context。

除了emptyCtx，其他类型的 Context 都依附在上级 Context 上

经过这个结构设计，如果要在整个任务链路上取消某个cancelCtx时，就能做到既取消自己，也把下级所有的cancelCtx都取消掉，同时还不会影响到上级和同级的其他节点。

我们让每个 goroutine 都携带了 Context ，那些做子任务的goroutine只要监听了这些子 cancelCtx 也就能收到信号，结束自己的运行，即通过Context 完成上级goroutine对下级 goroutine 的取消控制。

面对不同层级的goroutine的取消条件不同的情况，代码里只需要监听传递到 goroutine 里的 Context 就能做到，免除了监听多个信号的繁琐

针对Context的使用建议，Go官方提到了下面几点：

1. 不要将 Context 塞到结构体里。直接将 Context 类型作为函数的第一参数，而且一般都命名为 ctx。
2. 不要向函数传入一个 nil 的 context，如果你实在不知道传什么，标准库的TODO方法给你准备好了一个 emptyCtx。
   • 一般在初始化：用Background和TODO（只是用于不同场景下）：
     • Background通常被用于主函数、初始化以及测试中，作为一个顶层的context，也就是说一般我们创建的context都是基于Background
     • TODO是在不确定使用什么context的时候才会使用
3. 不要把本应该作为函数参数的类型塞到 context 中，context 存储的应该是一些在 goroutine 共享的数据，比如Server的信息等等

---

context的使用

初始化：（首先就是要定义根节点）

Go内置两个函数：Background()和TODO()，这两个函数分别返回一个实现了Context接口的background和todo。

我们代码中最开始都是以这两个内置的上下文对象作为最顶层的partent context，衍生出更多的子上下文对象。

• Background()主要用于main函数、初始化以及测试代码中，作为Context这个树结构的最顶层的Context，也就是根Context。
• TODO()，它目前还不知道具体的使用场景，如果我们不知道该使用什么Context的时候，可以使用这个。

background和todo本质上都是emptyCtx结构体类型，是一个不可取消，没有设置截止时间，没有携带任何值的Context。

with四系列

WithCancel

func WithCancel(parent Context) (ctx Context, cancel CancelFunc)
1

WithCancel返回带有新Done通道的父节点的副本。当调用返回的cancel函数或当关闭父上下文的Done通道时，将关闭返回上下文的Done通道，无论先发生什么情况。

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

func gen(ctx context.Context) <-chan int {
		dst := make(chan int)
		n := 1
		go func() {
			for {
				select {
				case <-ctx.Done():
					return // return结束该goroutine，防止泄露
				case dst <- n:
					n++
				}
			}
		}()
		return dst
	}
func main() {
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel() // 当我们取完需要的整数后调用cancel

	for n := range gen(ctx) {
		fmt.Println(n)
		if n == 5 {
			break
		}
	}
}
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

代码解释：

gen函数在单独的goroutine中生成整数并将它们发送到返回的通道。 gen的调用者在使用生成的整数之后需要取消上下文，以免gen启动的内部goroutine发生泄漏。

WithDeadline

func WithDeadline(parent Context, deadline time.Time) (Context, CancelFunc)
1

• 返回父上下文的副本，并将deadline调整为不迟于d(设置的时间)。
• 如果父上下文的deadline已经早于d（设置的时间），则WithDeadline(parent, d)在语义上等同于父上下文。
• 当截止日过期时，当调用返回的cancel函数时，或者当父上下文的Done通道关闭时，返回上下文的Done通道将被关闭，以最先发生的情况为准。

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

func main() {
	d := time.Now().Add(50 * time.Millisecond)
	ctx, cancel := context.WithDeadline(context.Background(), d)

	// 尽管ctx会过期，但在任何情况下调用它的cancel函数都是很好的实践。
	// 如果不这样做，可能会使上下文及其父类存活的时间超过必要的时间。
	defer cancel()

	select {
	case <-time.After(1 * time.Second):
		fmt.Println("overslept")
	case <-ctx.Done():
		fmt.Println(ctx.Err())
	}
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

代码解释

定义了一个50毫秒之后过期的deadline，然后我们调用context.WithDeadline(context.Background(), d)得到一个上下文（context）和一个取消函数（cancel），然后使用一个select让主程序陷入等待：等待1秒后打印overslept退出或者等待ctx过期后退出。

代码因为ctx 50毫秒后就会过期，所以ctx.Done()会先接收到context到期通知，并且会打印ctx.Err()的内容

WithTimeout

WithTimeout返回WithDeadline(parent, time.Now().Add(timeout))

func WithTimeout(parent Context, timeout time.Duration) (Context, CancelFunc)
1

取消此上下文将释放与其相关的资源，因此代码应该在此上下文中运行的操作完成后立即调用cancel，通常用于数据库或者网络连接的超时控制

// context.WithTimeout

var wg sync.WaitGroup

func worker(ctx context.Context) {
LOOP:
	for {
		fmt.Println("db connecting ...")
		time.Sleep(time.Millisecond * 10) // 假设正常连接数据库耗时10毫秒
		select {
		case <-ctx.Done(): // 50毫秒后自动调用
			break LOOP
		default:
		}
	}
	fmt.Println("worker done!")
	wg.Done()
}

func main() {
	// 设置一个50毫秒的超时
	ctx, cancel := context.WithTimeout(context.Background(), time.Millisecond*50)
	wg.Add(1)
	go worker(ctx)
	time.Sleep(time.Second * 5)
	cancel() // 通知子goroutine结束
	wg.Wait()
	fmt.Println("over")
}
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

WithValue

WithValue 返回与 key 关联的值为 val 的 parent 副本

func WithValue(parent Context, key, val interface{}) Context
1

WithValue返回父节点的副本，其中与key关联的值为val。仅对API和进程间传递请求域的数据使用上下文值，而不是使用它来传递可选参数给函数。

所提供的键必须是可比较的，并且不应该是string类型或任何其他内置类型，以避免使用上下文在包之间发生冲突。WithValue的用户应该为键定义自己的类型。为了避免在分配给interface{}时进行分配，上下文键通常具有具体类型struct{}。或者，导出的上下文关键变量的静态类型应该是指针或接口

// context.WithValue

type TraceCode string

var wg sync.WaitGroup

func worker(ctx context.Context) {
	key := TraceCode("TRACE_CODE")
	traceCode, ok := ctx.Value(key).(string) // 在子goroutine中获取trace code
	if !ok {
		fmt.Println("invalid trace code")
	}
LOOP:
	for {
		fmt.Printf("worker, trace code:%s\n", traceCode)
		time.Sleep(time.Millisecond * 10) // 假设正常连接数据库耗时10毫秒
		select {
		case <-ctx.Done(): // 50毫秒后自动调用
			break LOOP
		default:
		}
	}
	fmt.Println("worker done!")
	wg.Done()
}

func main() {
	// 设置一个50毫秒的超时
	ctx, cancel := context.WithTimeout(context.Background(), time.Millisecond*50)
	// 在系统的入口中设置trace code传递给后续启动的goroutine实现日志数据聚合
	ctx = context.WithValue(ctx, TraceCode("TRACE_CODE"), "12512312234")
	wg.Add(1)
	go worker(ctx)
	time.Sleep(time.Second * 5)
	cancel() // 通知子goroutine结束
	wg.Wait()
	fmt.Println("over")
}
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

最后在强调一下：

• 推荐以参数的方式显示传递Context
• 以Context作为参数的函数方法，应该把Context作为第一个参数。
• 给一个函数方法传递Context的时候，不要传递nil，如果不知道传递什么，就使用context.TODO()
• Context的Value相关方法应该传递请求域的必要数据，不应该用于传递可选参数
• Context是线程安全的，可以放心的在多个goroutine中传递