Go Context 使用

本文先讲解在什么情况下需要什么用 Go 的 Context 作为切入点来了解 Go 的 Context~

如何通知 goroutine 停止执行?

func main(){
	go func() { // 可以是任何签名的函数

		// 有什么方法，当我们需要主动的通知这个 goroutine 结束，
		// 有什么方法，给这个 goroutine 设置一个过期时间，等等等~
	}()
}

在解决这个问题前，先走点弯路，看看怎么等待 goroutine 完成执行~

等待 goroutine 完成执行

func main() {
    var wg sync.WaitGroup // WaitGroup 等待一组处理程序完成
    
	wg.Add(2)             // 将增量添加到 WaitGroup 的计数器中~

	go func() {
		time.Sleep(time.Second * 3)
		fmt.Println("1号完成任务~")
		wg.Done()        // 完成的话，将 WaitGroup 计数器逐一递减
	}()

	go func() {
		time.Sleep(time.Second * 2)
		fmt.Println("2号完成任务~")
		wg.Done()
	}()

	wg.Wait()           // 等待，直到 WaitGroup 计数器为零。

	fmt.Println("好了，大家都干完了，下班~")
}

Channel 通知

func main() {

	stop := make(chan struct{}) //

	go func(name string) {
		for {
			select {
			case <-stop:
				fmt.Println(name, " 收到通知，停止监控~")
				return
			default:
				fmt.Println(name, " 正在监控中...")
				time.Sleep(time.Second * 2)
			}
		}
	}("守望者")

	go func(name string) {
		time.Sleep(time.Second * 10)

		fmt.Println(name, " 通知监控停止~")
		stop <- struct{}{}

		time.Sleep(time.Second * 5)
	}("地球联盟")

	select {} // 骚操作，一直阻塞，一般用于写 demo 代码~
}

使用 chan + select 的方式，的确可以做到通知一个 goroutine 停止，
但如果有多个 goroutine，并且有嵌套，那么是否有比较好的方式来实现停止操作？

func main() {
	go func() {
		go func(){}()
		go func() {
			go func() {}()
			go func() {}()
		}()
	}()

	go func() {
		go func() {}()
		go func() {}()
		go func() {}()
	}()
	//  怎么优雅地让所有的 goroutine 都停止呢？如果用 chan，那么也太麻烦了吧~
}

Go Context

func main() {
    ctx, cancel := context.WithCancel(context.Background())
    
	go watch(ctx, "A")
	go watch(ctx, "B")
	go team_watch(ctx, "T")

	go func() {
		time.Sleep(time.Second * 10)

		fmt.Println("通知监控停止~")
		cancel()

		time.Sleep(time.Second * 5)
	}()

	select {}
}

func watch(ctx context.Context, name string) {
	for {
		select {
		case <-ctx.Done():
			fmt.Println(name, " 收到通知，停止监控~")
			return
		default:
			fmt.Println(name, " 正在监控中...")
			time.Sleep(time.Second * 3)
		}
	}
}

func team_watch(ctx context.Context, name string){
	go watch(ctx,name+"-a")
	go watch(ctx,name+"-b")
}

Context 的定义

type Context interface {

	// 获取该 Context 是否有设置截止时间，如果有，那么截止时间是什么时候
	Deadline() (deadline time.Time, ok bool)

	// 如果该方法返回的 chan 可以读取，则意味着 parent context 已经发起了取消请求
	Done() <-chan struct{}

	// 返回取消的错误原因，是什么原因 Context 被取消
	Err() error

	// 获取在该 Context 上绑定的值，这个值一般是线程安全的
	Value(key interface{}) interface{}
}

Done() + Err() 函数的一般使用方式

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:
        }
    }
}

Context 的实现

空实现 emptyCtx

// 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.
// 它不是struct {}，因为此类型的var必须具有不同的地址？？？
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)
)


// 它通常用于 main 函数，初始化，测试用例，或者最顶层的新请求
func Background() Context {
	return background
}

// 如果我们不知道该使用什么 Context 的时候，可以使用这个
func TODO() Context {
	return todo
}

cancelCtx，timerCtx，valueCtx 实现

func main() {
	// ctx, cancel := context.WithCancel(context.Background())
	ctxPartent := context.Background()

	// 创建出一个 cancelCtx 类型的子 Context，调用 cancel 函数通知结束
	ctxWithCancel, cancelWithCancel := context.WithCancel(ctxPartent)

	// 创建出一个 timerCtx 类型的子 Context
	deadline := time.Now() // 具体指定到某个时间点结束，或者调用 cancel 函数通知结束
	ctxWithDeadline, cancelWithDeadline := context.WithDeadline(ctxPartent, deadline)

	// 创建出一个 timerCtx 类型的子 Context
	timeout := time.Second * 3 // 从创建开始过多少时间结束，或者调用 cancel 函数通知结束
	ctxWithDeadline, cancelWithDeadline := context.WithTimeout(ctxPartent, timeout)

	// 创建出一个 valueCtx 类型的子 Context，没有 cancel 函数
	ctxWithValue := context.WithValue(ctxPartent, "Key", "Hello World")
}

cancelCtx 以及 timerCtx 还实现了另外一个接口 canceler

// 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{}
}

cancelCtx 详解

cancelCtx 的定义

// A cancelCtx can be canceled. When canceled, it also cancels any children that implement canceler.
// cancelCtx 可以被取消。 取消后，它也会调用所有实现 canceler接口 的子级
type cancelCtx struct {
	Context // parent context

	mu       sync.Mutex            // protects following fields 保护下面的字段
	done     chan struct{}         // created lazily, closed by first cancel call 延迟创建，在第一次调用 cancel 时，将其关闭
	children map[canceler]struct{} // set to nil by the first cancel call 第一次调用 cancel 时，将其设为 nil
	err      error                 // set to non-nil by the first cancel call  第一次调用 cancel 时，将其设为 non-nil
}

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       // 为啥不直接 return c.done ？？
}                  // 为了保护 c.done，不然外部通过 & 获取变量内存地址，就可以修改到 cancelCtx 内部的私有变量了

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

// cancel closes c.done, cancels each of c's children, 
// and, if removeFromParent is true, removes c from its parent's children.
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 // closedchan is a reusable closed channel; 可以重复使用(赋值)的已关闭通道
	} else {
		close(c.done)
	}
	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 {
		removeChild(c.Context, c)
	}
}

cancelCtx 的构造

func WithCancel(parent Context) (ctx Context, cancel CancelFunc) {
	c := newCancelCtx(parent)
	propagateCancel(parent, &c) // 调用 cancel 会关闭当前上下文，
	                            // 但也需要实现当 父上下文调用 cancel 时，也要关闭当前上下文
	return &c, func() { c.cancel(true, Canceled) }
	// ctxWithCancel, cancelWithCancel := context.WithCancel(ctxPartent)
	// 当调用 cancelWithCancel() 时，实际上是调用 Context 的 cancel 函数
	// 参数 Canceled 是这么初始化的 var Canceled = errors.New("context canceled") 
}

// newCancelCtx returns an initialized cancelCtx.
func newCancelCtx(parent Context) cancelCtx {
	return cancelCtx{Context: parent}
}

// propagateCancel arranges for child to be canceled when parent is.
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
			// 如果父上下文已经 cancel 了，那么子也直接 cancel
			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():
			}
		}()
	}
}

// parentCancelCtx follows a chain of parent references until it finds a *cancelCtx.
// This function understands how each of the concrete types in this package represents its parent.
// parentCancelCtx 循父引用链，直到找到 _cancelCtx。此函数了解此包中的每个具体类型如何表示其父类型。
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 // 赋值后，还是在 for 循环中，类似递归~
		default:
			return nil, false
		}
	}
}