您好,登錄后才能下訂單哦!
本篇內容主要講解“golang協程關閉實例分析”,感興趣的朋友不妨來看看。本文介紹的方法操作簡單快捷,實用性強。下面就讓小編來帶大家學習“golang協程關閉實例分析”吧!
結合如下典型場景,主進程中起多個協程,這些協程會
1.共同消費一個數據通道 data channel
2.也可能共享一個退出通道channel或context
那么,應該如何正確關閉呢
并不推薦強制停止,更多的時候我們希望在停止時,干一點事比如資源清理/連接清理等,這時候最好的方式就是通知協程退出,具體何時退出和退出前做什么完全由當前要關閉的協程控制。
通知一般有三種方式
適用簡單任務,復雜的更推薦context單獨通知
// cancelFn 數據通道關閉通知退出 func cancelFn(dataChan chan int) { for { select { case val, ok := <-dataChan: // 關閉data通道時,通知退出 // 一個可選是判斷data=指定值時退出 if !ok { log.Printf("Channel closed !!!") return } log.Printf("Revice dataChan %d\n", val) } } }
部分簡單場景適用
// exitChannelFn 單獨退出通道關閉通知退出 func exitChannelFn(wg *sync.WaitGroup, taskNo int, dataChan chan int, exitChan chan struct{}) { defer wg.Done() for { select { case val, ok := <-dataChan: if !ok { log.Printf("Task %d channel closed !!!", taskNo) return } log.Printf("Task %d revice dataChan %d\n", taskNo, val) // 關閉exit通道時,通知退出 case <-exitChan: log.Printf("Task %d revice exitChan signal!\n", taskNo) return } } }
主流推薦
// contextCancelFn context取消或超時通知退出 func contextCancelFn(wg *sync.WaitGroup, taskNo int, dataChan chan int, ctx context.Context) { defer wg.Done() for { select { case val, ok := <-dataChan: if !ok { log.Printf("Task %d channel closed !!!", taskNo) return } log.Printf("Task %d revice dataChan %d\n", taskNo, val) // ctx取消或超時,通知退出 case <-ctx.Done(): log.Printf("Task %d revice exit signal!\n", taskNo) return } } }
go func可以立即創建一個協程,因此常常遇到我們可能在任何一個地方創建協程,但是在哪里關閉呢,是需要統一管理嗎?官方推薦的最佳實踐就是,誰負責創建協程誰負責關閉協程。
參考如下,每次調用execDataTaskFunc函數執行都會起一個協程異步執行,協程關閉通過監控外層函數context參數來實現。
func execDataTaskFunc(ctx context.Context, dataChan chan int, taskName string) chan int { out := make(chan int) log.Printf("Task %s start!\n", taskName) go func() { defer close(out) for { select { case data, ok := <-dataChan: if !ok { log.Printf("Task %s revice data channel close signal!\n", taskName) return } // do something out <- data case <-ctx.Done(): log.Printf("Task %s revice exit signal!\n", taskName) return } } }() return out }
通常對于正在運行的協程,發出退出通知后,具體程序何時才能退出呢?一般如下三種方式
最常用,參考如下
// 多個任務并行控制,等待所有任務完成 func TestTaskControl(t *testing.T) { dataChan := make(chan int) taskNum := 3 wg := sync.WaitGroup{} wg.Add(taskNum) // 起多個協程,data關閉時退出 for i := 0; i < taskNum; i++ { go func(taskNo int) { defer wg.Done() t.Logf("Task %d run\n", taskNo) for { select { case _, ok := <-dataChan: if !ok { t.Logf("Task %d notify to stop\n", taskNo) return } } } }(i) } // 通知退出 go func() { time.Sleep(3 * time.Second) close(dataChan) }() // 等待退出完成 wg.Wait() }
參考如下,對于部分任務場景,協程數據輸出到新建的channel中,可以在此channel上阻塞等待,直到協程通知關閉時,關閉此channel然后程序退出。
// 多個任務并行控制,等待所有任務完成 func TestTaskControl2(t *testing.T) { dataChan := make(chan int) // 起協程返回新chan,在輸出chan等待判斷完成 out := make(chan int) go func() { defer close(out) // 結束則自動關閉 for { select { case _, ok := <-dataChan: if !ok { t.Logf("Task notify to stop\n") return } } } }() // 通知退出 go func() { time.Sleep(3 * time.Second) close(dataChan) }() dataChan <- 1 // 等待退出完成 for data := range out { t.Logf("%d\n", data) } }
對于部分任務,能夠估算從通知關閉到實際關閉時間,則可等待足夠長時間來保證協程關閉然后退出,實際場景并不推薦,帶有一定不確定性,很容易出錯。
func TestTaskControl3(t *testing.T) { dataChan := make(chan int) // 起協程返回新chan out := make(chan int) go func() { defer close(out) // 結束則自動關閉 for { select { case _, ok := <-dataChan: if !ok { t.Logf("Task notify to stop\n") return } } } }() // 通知退出 go func() { time.Sleep(3 * time.Second) close(dataChan) }() dataChan <- 1 // 等待足夠長時間,退出完成 time.Sleep(10 * time.Second) }
結合三大原則,這里展示部分復雜場景的協程關閉方案。
如下是多個任務執行,每個任務一個協程,現在考慮如下目標
支持多級嵌套,父任務停止后,子任務自動停止
方案:使用context通知,WaitGroup等待所有任務關閉后退出
任務運行代碼
type TaskFunc func(ctx context.Context) func runTaskFunc(wg *sync.WaitGroup, ctx context.Context, taskName string, f TaskFunc) { defer wg.Done() log.Printf("Task %s start!\n", taskName) f(ctx) for { select { case <-ctx.Done(): log.Printf("Task %s revice exit signal!\n", taskName) return } } }
整體實現代碼
// 簡單并行任務-同時停止 func TestStop(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) var wg = sync.WaitGroup{} // 起多個任務 wg.Add(1) go runTaskFunc(&wg, ctx, "A", func(ctx context.Context) { wg.Add(1) go runTaskFunc(&wg, ctx, "B", func(ctx context.Context) { wg.Add(1) go runTaskFunc(&wg, ctx, "C", func(ctx context.Context) { wg.Add(1) go runTaskFunc(&wg, ctx, "D", func(ctx context.Context) {}) }) }) wg.Add(1) go runTaskFunc(&wg, ctx, "E", func(ctx context.Context) { wg.Add(1) go runTaskFunc(&wg, ctx, "F", func(ctx context.Context) { wg.Add(1) go runTaskFunc(&wg, ctx, "G", func(ctx context.Context) {}) }) }) }) // 通知關閉 go func() { time.Sleep(3 * time.Second) cancel() }() // 等待全部關閉后退出 wg.Wait() }
協程關閉是無序的,如下
2023/01/07 22:40:09 Task A start!
2023/01/07 22:40:09 Task E start!
2023/01/07 22:40:09 Task F start!
2023/01/07 22:40:09 Task G start!
2023/01/07 22:40:09 Task B start!
2023/01/07 22:40:09 Task C start!
2023/01/07 22:40:09 Task D start!
2023/01/07 22:40:12 Task A revice exit signal!
2023/01/07 22:40:12 Task G revice exit signal!
2023/01/07 22:40:12 Task B revice exit signal!
2023/01/07 22:40:12 Task F revice exit signal!
2023/01/07 22:40:12 Task D revice exit signal!
2023/01/07 22:40:12 Task C revice exit signal!
2023/01/07 22:40:12 Task E revice exit signal!
還是上述場景,現在需求是:控制停止順序,先停EFG 再停BCD 最后停A
方案:借助context通知,指定多個cancel點,WaitGroup等待所有任務關閉后退出
// 簡單并行任務-控制停止順序 func TestStop2(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) ctxb, cancelb := context.WithCancel(ctx) ctxe, cancele := context.WithCancel(ctx) var wg = sync.WaitGroup{} // 起多個任務 wg.Add(1) go runTaskFunc(&wg, ctx, "A", func(ctx context.Context) { wg.Add(1) go runTaskFunc(&wg, ctxb, "B", func(ctx context.Context) { wg.Add(1) go runTaskFunc(&wg, ctx, "C", func(ctx context.Context) { wg.Add(1) go runTaskFunc(&wg, ctx, "D", func(ctx context.Context) {}) }) }) wg.Add(1) go runTaskFunc(&wg, ctxe, "E", func(ctx context.Context) { wg.Add(1) go runTaskFunc(&wg, ctx, "F", func(ctx context.Context) { wg.Add(1) go runTaskFunc(&wg, ctx, "G", func(ctx context.Context) {}) }) }) }) // 通知關閉 go func() { time.Sleep(3 * time.Second) cancele() time.Sleep(3 * time.Second) cancelb() time.Sleep(3 * time.Second) cancel() }() // 等待全部關閉后退出 wg.Wait() }
運行如下,協程按照指定順序關閉
2023/01/07 22:40:40 Task A start!
2023/01/07 22:40:40 Task E start!
2023/01/07 22:40:40 Task F start!
2023/01/07 22:40:40 Task G start!
2023/01/07 22:40:40 Task B start!
2023/01/07 22:40:40 Task C start!
2023/01/07 22:40:40 Task D start!
2023/01/07 22:40:43 Task E revice exit signal!
2023/01/07 22:40:43 Task F revice exit signal!
2023/01/07 22:40:43 Task G revice exit signal!
2023/01/07 22:40:46 Task B revice exit signal!
2023/01/07 22:40:46 Task D revice exit signal!
2023/01/07 22:40:46 Task C revice exit signal!
2023/01/07 22:40:49 Task A revice exit signal!
考慮如下場景,A->B->C嵌套起協程,每個協程創建新的channel傳輸數據給下游
如下起任務,每個任務可以通過context或者data channel關閉來通知退出
func execDataTaskFunc(ctx context.Context, dataChan chan int, taskName string) chan int { out := make(chan int) //out := make(chan int, 100) log.Printf("Task %s start!\n", taskName) go func() { defer close(out) for { select { case data, ok := <-dataChan: if !ok { log.Printf("Task %s revice data channel close signal!\n", taskName) return } time.Sleep(2 * time.Second) out <- data case <-ctx.Done(): log.Printf("Task %s revice exit signal!\n", taskName) return } } }() return out }
整體流程如下
func TestDataTaskStop(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) defer cancel() dataChanInput := make(chan int) // 嵌套運行協程 taskChanA := execDataTaskFunc(ctx, dataChanInput, "A") taskChanB := execDataTaskFunc(ctx, taskChanA, "B") taskChanC := execDataTaskFunc(ctx, taskChanB, "C") // 通知退出 go func() { i := 0 for { select { case <-time.After(time.Second): i = i + 1 if i == 10 { t.Logf("Notify to stop!!!") close(dataChanInput) //cancel() return } dataChanInput <- i } } }() // 等待退出 for data := range taskChanC { t.Logf("Out->%d", data) } }
這里數據每條數據產生間隔1秒,每個任務處理時長為2秒,也就是說通知關閉時,可能上游任務處理中,下游還沒來得及處理,因此期望的是逐級依次關閉A/B/C,確保上游數據處理完成傳給下游,不要丟失數據。
對比context通知退出和data channel關閉通知退出,對比如下。可以看到如果我們是有中間處理和逐級關閉需求的還是要依賴close關閉協程來通知,context全局通知退出是無序的,無法保證數據不丟失。
cancel()-context通知退出
執行如下,A/B/C同時退出,數據出現丟失
2023/01/07 23:23:59 Task A start!
2023/01/07 23:23:59 Task B start!
2023/01/07 23:23:59 Task C start!
complex_test.go:174: Out->1
complex_test.go:174: Out->2
complex_test.go:174: Out->3
complex_test.go:174: Out->4
complex_test.go:174: Out->5
complex_test.go:174: Out->6
complex_test.go:161: Notify to stop!!!
2023/01/07 23:24:18 Task C revice exit signal!
complex_test.go:174: Out->7
close(dataChanInput)通知退出
執行如下,A/B/C逐級依次關閉,數據沒有丟失
2023/01/07 23:20:18 Task A start!
2023/01/07 23:20:18 Task B start!
2023/01/07 23:20:18 Task C start!
complex_test.go:174: Out->1
complex_test.go:174: Out->2
complex_test.go:174: Out->3
complex_test.go:174: Out->4
complex_test.go:174: Out->5
complex_test.go:174: Out->6
complex_test.go:161: Notify to stop!!!
complex_test.go:174: Out->7
2023/01/07 23:20:37 Task A revice data channel close signal!
complex_test.go:174: Out->8
2023/01/07 23:20:39 Task B revice data channel close signal!
2023/01/07 23:20:41 Task C revice data channel close signal!
complex_test.go:174: Out->9
到此,相信大家對“golang協程關閉實例分析”有了更深的了解,不妨來實際操作一番吧!這里是億速云網站,更多相關內容可以進入相關頻道進行查詢,關注我們,繼續學習!
免責聲明:本站發布的內容(圖片、視頻和文字)以原創、轉載和分享為主,文章觀點不代表本網站立場,如果涉及侵權請聯系站長郵箱:is@yisu.com進行舉報,并提供相關證據,一經查實,將立刻刪除涉嫌侵權內容。