溫馨提示×

您好，登錄后才能下訂單哦！

密碼登錄×
登錄注冊(cè)×
其他方式登錄
點(diǎn)擊 登錄注冊(cè) 即表示同意《億速云用戶服務(wù)條款》

golang協(xié)程關(guān)閉實(shí)例分析

發(fā)布時(shí)間：2023-03-17 11:50:48 來源：億速云 閱讀：162 作者：iii 欄目：開發(fā)技術(shù)

本篇內(nèi)容主要講解“golang協(xié)程關(guān)閉實(shí)例分析”，感興趣的朋友不妨來看看。本文介紹的方法操作簡(jiǎn)單快捷，實(shí)用性強(qiáng)。下面就讓小編來帶大家學(xué)習(xí)“golang協(xié)程關(guān)閉實(shí)例分析”吧!

場(chǎng)景

結(jié)合如下典型場(chǎng)景，主進(jìn)程中起多個(gè)協(xié)程，這些協(xié)程會(huì)

1.共同消費(fèi)一個(gè)數(shù)據(jù)通道 data channel

2.也可能共享一個(gè)退出通道channel或context

golang協(xié)程關(guān)閉實(shí)例分析

那么，應(yīng)該如何正確關(guān)閉呢

原則1-協(xié)程接受通知主動(dòng)關(guān)閉

并不推薦強(qiáng)制停止，更多的時(shí)候我們希望在停止時(shí)，干一點(diǎn)事比如資源清理/連接清理等，這時(shí)候最好的方式就是通知協(xié)程退出，具體何時(shí)退出和退出前做什么完全由當(dāng)前要關(guān)閉的協(xié)程控制。

通知一般有三種方式

data channel關(guān)閉通知退出

適用簡(jiǎn)單任務(wù)，復(fù)雜的更推薦context單獨(dú)通知

// cancelFn 數(shù)據(jù)通道關(guān)閉通知退出
func cancelFn(dataChan chan int) {
	for {
		select {
		case val, ok := <-dataChan:
			// 關(guān)閉data通道時(shí)，通知退出
			// 一個(gè)可選是判斷data=指定值時(shí)退出
			if !ok {
				log.Printf("Channel closed ！?。?quot;)
				return
			}

			log.Printf("Revice dataChan %d\n", val)
		}
	}
}

exit channel關(guān)閉通知退出

部分簡(jiǎn)單場(chǎng)景適用

// exitChannelFn 單獨(dú)退出通道關(guān)閉通知退出
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)

			// 關(guān)閉exit通道時(shí)，通知退出
		case <-exitChan:
			log.Printf("Task %d  revice exitChan signal!\n", taskNo)
			return
		}
	}

}

context超時(shí)或取消通知退出

主流推薦

// contextCancelFn context取消或超時(shí)通知退出
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取消或超時(shí)，通知退出
		case <-ctx.Done():
			log.Printf("Task %d  revice exit signal!\n", taskNo)
			return
		}
	}

}

原則2-誰負(fù)責(zé)創(chuàng)建協(xié)程誰負(fù)責(zé)關(guān)閉協(xié)程

go func可以立即創(chuàng)建一個(gè)協(xié)程，因此常常遇到我們可能在任何一個(gè)地方創(chuàng)建協(xié)程，但是在哪里關(guān)閉呢，是需要統(tǒng)一管理嗎？官方推薦的最佳實(shí)踐就是，誰負(fù)責(zé)創(chuàng)建協(xié)程誰負(fù)責(zé)關(guān)閉協(xié)程。

參考如下，每次調(diào)用execDataTaskFunc函數(shù)執(zhí)行都會(huì)起一個(gè)協(xié)程異步執(zhí)行，協(xié)程關(guān)閉通過監(jiān)控外層函數(shù)context參數(shù)來實(shí)現(xiàn)。

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
}

原則3-等待所有協(xié)程關(guān)閉再退出

通常對(duì)于正在運(yùn)行的協(xié)程，發(fā)出退出通知后，具體程序何時(shí)才能退出呢？一般如下三種方式

WaitGroup/ErrGroup判斷所有協(xié)程關(guān)閉后退出

最常用，參考如下

// 多個(gè)任務(wù)并行控制，等待所有任務(wù)完成
func TestTaskControl(t *testing.T) {
	dataChan := make(chan int)

	taskNum := 3

	wg := sync.WaitGroup{}
	wg.Add(taskNum)

	// 起多個(gè)協(xié)程，data關(guān)閉時(shí)退出
	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關(guān)閉后退出

參考如下，對(duì)于部分任務(wù)場(chǎng)景，協(xié)程數(shù)據(jù)輸出到新建的channel中，可以在此channel上阻塞等待，直到協(xié)程通知關(guān)閉時(shí)，關(guān)閉此channel然后程序退出。

// 多個(gè)任務(wù)并行控制，等待所有任務(wù)完成
func TestTaskControl2(t *testing.T) {
	dataChan := make(chan int)

	// 起協(xié)程返回新chan，在輸出chan等待判斷完成
	out := make(chan int)
	go func() {
		defer close(out) // 結(jié)束則自動(dòng)關(guān)閉

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

等待足夠長(zhǎng)時(shí)間后關(guān)閉

對(duì)于部分任務(wù)，能夠估算從通知關(guān)閉到實(shí)際關(guān)閉時(shí)間，則可等待足夠長(zhǎng)時(shí)間來保證協(xié)程關(guān)閉然后退出，實(shí)際場(chǎng)景并不推薦，帶有一定不確定性，很容易出錯(cuò)。

func TestTaskControl3(t *testing.T) {
	dataChan := make(chan int)

	// 起協(xié)程返回新chan
	out := make(chan int)
	go func() {
		defer close(out) // 結(jié)束則自動(dòng)關(guān)閉

		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

	// 等待足夠長(zhǎng)時(shí)間，退出完成
	time.Sleep(10 * time.Second)
}

復(fù)雜退出場(chǎng)景

結(jié)合三大原則，這里展示部分復(fù)雜場(chǎng)景的協(xié)程關(guān)閉方案。

嵌套協(xié)程，同時(shí)關(guān)閉

如下是多個(gè)任務(wù)執(zhí)行，每個(gè)任務(wù)一個(gè)協(xié)程，現(xiàn)在考慮如下目標(biāo)

支持多級(jí)嵌套，父任務(wù)停止后，子任務(wù)自動(dòng)停止

golang協(xié)程關(guān)閉實(shí)例分析

方案：使用context通知，WaitGroup等待所有任務(wù)關(guān)閉后退出

任務(wù)運(yùn)行代碼

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

}

整體實(shí)現(xiàn)代碼

// 簡(jiǎn)單并行任務(wù)-同時(shí)停止
func TestStop(t *testing.T) {
	ctx, cancel := context.WithCancel(context.Background())

	var wg = sync.WaitGroup{}

	// 起多個(gè)任務(wù)
	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) {})
			})
		})
	})

	// 通知關(guān)閉
	go func() {
		time.Sleep(3 * time.Second)
		cancel()
	}()

	// 等待全部關(guān)閉后退出
	wg.Wait()
}

協(xié)程關(guān)閉是無序的，如下

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!

嵌套協(xié)程，指定順序關(guān)閉

還是上述場(chǎng)景，現(xiàn)在需求是：控制停止順序，先停EFG 再停BCD 最后停A

golang協(xié)程關(guān)閉實(shí)例分析

方案：借助context通知，指定多個(gè)cancel點(diǎn)，WaitGroup等待所有任務(wù)關(guān)閉后退出

// 簡(jiǎn)單并行任務(wù)-控制停止順序
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{}

	// 起多個(gè)任務(wù)
	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) {})
			})
		})
	})

	// 通知關(guān)閉
	go func() {
		time.Sleep(3 * time.Second)
		cancele()
		time.Sleep(3 * time.Second)
		cancelb()
		time.Sleep(3 * time.Second)
		cancel()
	}()

	// 等待全部關(guān)閉后退出
	wg.Wait()
}

運(yùn)行如下，協(xié)程按照指定順序關(guān)閉

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!

嵌套協(xié)程，逐級(jí)關(guān)閉

考慮如下場(chǎng)景，A->B->C嵌套起協(xié)程，每個(gè)協(xié)程創(chuàng)建新的channel傳輸數(shù)據(jù)給下游

golang協(xié)程關(guān)閉實(shí)例分析

如下起任務(wù)，每個(gè)任務(wù)可以通過context或者data channel關(guān)閉來通知退出

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)

	// 嵌套運(yùn)行協(xié)程
	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)
	}
}

這里數(shù)據(jù)每條數(shù)據(jù)產(chǎn)生間隔1秒，每個(gè)任務(wù)處理時(shí)長(zhǎng)為2秒，也就是說通知關(guān)閉時(shí)，可能上游任務(wù)處理中，下游還沒來得及處理，因此期望的是逐級(jí)依次關(guān)閉A/B/C，確保上游數(shù)據(jù)處理完成傳給下游，不要丟失數(shù)據(jù)。

對(duì)比context通知退出和data channel關(guān)閉通知退出，對(duì)比如下?？梢钥吹饺绻覀兪?strong>有中間處理和逐級(jí)關(guān)閉需求的還是要依賴close關(guān)閉協(xié)程來通知，context全局通知退出是無序的，無法保證數(shù)據(jù)不丟失。

  • cancel()-context通知退出

執(zhí)行如下，A/B/C同時(shí)退出，數(shù)據(jù)出現(xiàn)丟失

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)通知退出

執(zhí)行如下，A/B/C逐級(jí)依次關(guān)閉，數(shù)據(jù)沒有丟失

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

到此，相信大家對(duì)“golang協(xié)程關(guān)閉實(shí)例分析”有了更深的了解，不妨來實(shí)際操作一番吧！這里是億速云網(wǎng)站，更多相關(guān)內(nèi)容可以進(jìn)入相關(guān)頻道進(jìn)行查詢，關(guān)注我們，繼續(xù)學(xué)習(xí)！

向AI問一下細(xì)節(jié)

免責(zé)聲明：本站發(fā)布的內(nèi)容（圖片、視頻和文字）以原創(chuàng)、轉(zhuǎn)載和分享為主，文章觀點(diǎn)不代表本網(wǎng)站立場(chǎng)，如果涉及侵權(quán)請(qǐng)聯(lián)系站長(zhǎng)郵箱：is@yisu.com進(jìn)行舉報(bào)，并提供相關(guān)證據(jù)，一經(jīng)查實(shí)，將立刻刪除涉嫌侵權(quán)內(nèi)容。

AI
助
手