Go context.WithCancel()的使用

塞北de雪 2023-08-16 發(fā)布于江蘇

展開全文

WithCancel可以將一個Context包裝為cancelCtx,并提供一個取消函數(shù),調用這個取消函數(shù),可以Cancel對應的Context

Go語言context包-cancelCtx^[1]

疑問

context.WithCancel()取消機制的理解^[2]

父母5s鐘后出門，倒計時，父母在時要學習，父母一走就可以玩

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.v/context#WithCancel:^[3]

// 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 副本。 返回的上下文的完成通道在調用返回的取消函數(shù)或父上下文的完成通道關閉時關閉，以先發(fā)生者為準。

//取消此上下文會釋放與其關聯(lián)的資源，因此代碼應在此上下文中運行的操作完成后立即調用取消。

func WithCancel(parent Context) (ctx Context, cancel CancelFunc) {
 if parent == nil {
  panic("cannot create context from nil parent")
 }
 c := newCancelCtx(parent)   // 將parent作為父節(jié)點context 生成一個新的子節(jié)點

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

WithCancel最后返回子上下文和一個cancelFunc函數(shù)，而cancelFunc函數(shù)里調用了cancelCtx這個結構體的方法cancel

(代碼基于go 1.16； 1.17有所改動)

// 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節(jié)點下的子節(jié)點
 err      error                 // set to non-nil by the first cancel call  err用于存儲錯誤信息 表示任務結束的原因
}

在cancelCtx這個結構體中，字段done是一個傳遞空結構體類型的channel，用來在上下文取消時關閉這個通道，err就是在上下文被取消時告訴用戶這個上下文取消了，可以用ctx.Err()來獲取信息

canceler是一個實現(xiàn)接口，用于Ctx的終止。實現(xiàn)該接口的Context有cancelCtx和timerCtx，而emptyCtx和valueCtx沒有實現(xiàn)該接口。

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


// 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的子節(jié)點
* 2、從父節(jié)點中移除該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關閉，用以發(fā)送關閉信號
 if c.done == nil {
  c.done = closedchan
 } else {
  close(c.done) // 注意這一步
 }

  // 將子節(jié)點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節(jié)點從父節(jié)點上移除
  removeChild(c.Context, c)
 }
}

對于cancel函數(shù)，其取消了基于該上下文的所有子上下文以及把自身從父上下文中取消

對于更多removeFromParent代碼分析,和其他Context的使用，強烈建議閱讀深入理解Golang之Context（可用于實現(xiàn)超時機制）^[4]

 // 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./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，類型為結構體。可用于監(jiān)聽當前 channel 是否已經被關閉。

Done()用來監(jiān)聽cancel操作(對于cancelCtx)或超時操作(對于timerCtx)，當執(zhí)行取消操作或超時時，c.done會被close，這樣就能從一個已經關閉的channel里一直獲取對應的零值，<-ctx.Done便不會再阻塞

(代碼基于go 1.16； 1.17有所改動)

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循環(huán)里select不斷嘗試從 <-ctx.Done()里讀取出內容，但此時并沒有任何給 c.done這個channel寫入數(shù)據的操作，（類似c.done <- struct{}{}）,故而在for循環(huán)里每次select時，這個case都不滿足條件，一直阻塞著。每次都執(zhí)行default代碼段

而在執(zhí)行cancelFunc時，在func (c *cancelCtx) cancel(removeFromParent bool, err error)里面，會有一個close(c.done)的操作。而從一個已經關閉的channel里可以一直獲取對應的零值，即 select可以命中，進入case res := <-ctx.Done():代碼段

可用如下代碼驗證：

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?。?！")
 }

}

輸出：

make(chan struct{})后不為nil！??！
I am working!
res2: {}
I am working!
I am working!
I am working!
I am working!
res: {}

而如果不向沒有緩存的cuiChan寫入數(shù)據，直接close，即

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

res2: {}
res2: {}
res2: {}
res2: {}
res2: {}
res2: {}
res2: {}
...
//一直打印下去

更多參考：

深入理解Golang之Context（可用于實現(xiàn)超時機制）^[5]

回答我，停止 Goroutine 有幾種方法？

golang context的done和cancel的理解 for循環(huán)channel實現(xiàn)context.Done()阻塞輸出^[6]

更多關于channel阻塞與close的代碼

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寫入第一條數(shù)據..."
 ch <- "向ch這個channel寫入第二條數(shù)據?。?！"

 close(ch) // 當channel被close后， v,ok 中的ok就會變?yōu)閒alse

 time.Sleep(10 * time.Second)
}

輸出為：

----開始----
<-ch一直沒有東西寫進去，會一直阻塞著,直到3秒鐘后


v,ok 向ch這個channel寫入第一條數(shù)據... true
----開始----
v,ok 向ch這個channel寫入第二條數(shù)據！?。?nbsp;true
----開始----
v,ok  false
結束

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, "次向通道中寫入數(shù)據"):

   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次向通道中寫入數(shù)據
接收到的值:  第2次向通道中寫入數(shù)據
接收到的值:  第3次向通道中寫入數(shù)據
接收到的值:  第4次向通道中寫入數(shù)據
接收到的值:  第5次向通道中寫入數(shù)據
接收到的值:  第6次向通道中寫入數(shù)據
接收到的值:  第7次向通道中寫入數(shù)據
接收到的值:  第8次向通道中寫入數(shù)據
接收到的值:  第9次向通道中寫入數(shù)據
接收到的值:  第10次向通道中寫入數(shù)據
接收到的值:  第11次向通道中寫入數(shù)據
接收到的值:  第12次向通道中寫入數(shù)據
接收到的值:  第13次向通道中寫入數(shù)據
接收到的值:  第14次向通道中寫入數(shù)據
接收到的值:  第15次向通道中寫入數(shù)據
接收到的值:  第16次向通道中寫入數(shù)據
接收到的值:  第17次向通道中寫入數(shù)據
接收到的值:  第18次向通道中寫入數(shù)據
接收到的值:  第19次向通道中寫入數(shù)據
接收到的值:  第20次向通道中寫入數(shù)據
接收到的值:  第21次向通道中寫入數(shù)據
接收到的值:  第22次向通道中寫入數(shù)據
接收到的值:  第23次向通道中寫入數(shù)據
接收到的值:  第24次向通道中寫入數(shù)據
接收到的值:  第25次向通道中寫入數(shù)據
接收到的值:  第26次向通道中寫入數(shù)據
接收到的值:  第27次向通道中寫入數(shù)據
接收到的值:  第28次向通道中寫入數(shù)據
接收到的值:  第29次向通道中寫入數(shù)據
接收到的值:  第30次向通道中寫入數(shù)據
接收到的值:  第31次向通道中寫入數(shù)據
結束