溫馨提示×
您好，登錄后才能下訂單哦！
點擊 登錄注冊 即表示同意《億速云用戶服務條款》
您好，登錄后才能下訂單哦！
本篇內容主要講解“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進行舉報，并提供相關證據，一經查實，將立刻刪除涉嫌侵權內容。
