您好,登錄后才能下訂單哦!
本篇內(nèi)容主要講解“Go語(yǔ)言中Goroutine退出機(jī)制如何使用”,感興趣的朋友不妨來(lái)看看。本文介紹的方法操作簡(jiǎn)單快捷,實(shí)用性強(qiáng)。下面就讓小編來(lái)帶大家學(xué)習(xí)“Go語(yǔ)言中Goroutine退出機(jī)制如何使用”吧!
goroutine的調(diào)度是由 Golang 運(yùn)行時(shí)進(jìn)行管理的。同一個(gè)程序中的所有 goroutine 共享同一個(gè)地址空間。goroutine設(shè)計(jì)的退出機(jī)制是由goroutine自己退出,不能在外部強(qiáng)制結(jié)束一個(gè)正在執(zhí)行的goroutine(只有一種情況正在運(yùn)行的goroutine會(huì)因?yàn)槠渌鹓oroutine的結(jié)束被終止,就是main函數(shù)退出或程序停止執(zhí)行)。下面我先介紹下幾種退出方式:
kill進(jìn)程/進(jìn)程crash
當(dāng)進(jìn)程被強(qiáng)制退出,所有它占有的資源都會(huì)還給操作系統(tǒng),而goroutine作為進(jìn)程內(nèi)的線程,資源被收回了,那么還未結(jié)束的goroutine也會(huì)直接退出
main函數(shù)結(jié)束
同理,當(dāng)主函數(shù)結(jié)束,goroutine的資源也會(huì)被收回,直接退出。具體可參考下下面的demo,其中g(shù)o routine里需要print出來(lái)的語(yǔ)句是永遠(yuǎn)也不會(huì)出現(xiàn)的。
package main import ( "fmt" "time" ) func routineTest() { time.Sleep(time.Second) fmt.Println("I'm alive") } func main(){ fmt.Println("start test") go routineTest() fmt.Println("end test") }
Go實(shí)現(xiàn)了兩種并發(fā)形式。第一種是大家普遍認(rèn)知的:多線程共享內(nèi)存。其實(shí)就是Java或者C++等語(yǔ)言中的多線程開發(fā)。另外一種是Go語(yǔ)言特有的,也是Go語(yǔ)言推薦的:CSP(communicating sequential processes)并發(fā)模型。CSP并發(fā)模型是在1970年左右提出的概念,屬于比較新的概念,不同于傳統(tǒng)的多線程通過(guò)共享內(nèi)存來(lái)通信,CSP講究的是“以通信的方式來(lái)共享內(nèi)存”。
其核心思想為:
DO NOT COMMUNICATE BY SHARING MEMORY; INSTEAD, SHARE MEMORY BY COMMUNICATING.
“不要以共享內(nèi)存的方式來(lái)通信,相反,要通過(guò)通信來(lái)共享內(nèi)存?!?/p>
普通的線程并發(fā)模型,就是像Java、C++、或者Python,他們線程間通信都是通過(guò)共享內(nèi)存的方式來(lái)進(jìn)行的。非常典型的方式就是,在訪問(wèn)共享數(shù)據(jù)(例如數(shù)組、Map、或者某個(gè)結(jié)構(gòu)體或?qū)ο螅┑臅r(shí)候,通過(guò)鎖來(lái)訪問(wèn),因此,在很多時(shí)候,衍生出一種方便操作的數(shù)據(jù)結(jié)構(gòu),叫做“線程安全的數(shù)據(jù)結(jié)構(gòu)”。例如Java提供的包”java.util.concurrent”中的數(shù)據(jù)結(jié)構(gòu)。Go中也實(shí)現(xiàn)了傳統(tǒng)的線程并發(fā)模型。
Go的CSP并發(fā)模型,就是通過(guò)goroutine和channel來(lái)實(shí)現(xiàn)的。
因?yàn)椴皇潜疚闹攸c(diǎn),在此對(duì)channel不做過(guò)多介紹,只需要了解channel是goroutine之間的通信機(jī)制。 通俗的講,就是各個(gè)goroutine之間通信的”管道“,有點(diǎn)類似于Linux中的管道。channel是go最推薦的goroutine間的通信方式,同時(shí)通過(guò)channel來(lái)通知goroutine退出也是最主要的goroutine退出方式。goroutine雖然不能強(qiáng)制結(jié)束另外一個(gè)goroutine,但是它可以通過(guò)channel通知另外一個(gè)goroutine你的表演該結(jié)束了。
package main import ( "fmt" "time" ) func cancelByChannel(quit <-chan time.Time) { for { select { case <-quit: fmt.Println("cancel goroutine by channel!") return default: fmt.Println("I'm alive") time.Sleep(1 * time.Second) } } } func main() { quit := time.After(time.Second * 10) go cancelByChannel(quit) time.Sleep(15*time.Second) fmt.Println("I'm done") }
在上面的例子中,我們用時(shí)間定義了一個(gè)channel,當(dāng)10秒后,會(huì)給到goroutine一個(gè)退出信號(hào),然后go routine就會(huì)退出。這樣我們就實(shí)現(xiàn)了在其他線程中通知另一個(gè)線程退出的功能。
通過(guò)channel通知goroutine退出還有一個(gè)更好的方法就是使用context。沒錯(cuò),就是我們?cè)谌粘i_發(fā)中接口通用的第一個(gè)參數(shù)context。它本質(zhì)還是接收一個(gè)channel數(shù)據(jù),只是是通過(guò)ctx.Done()獲取。將上面的示例稍作修改即可。
package main import ( "context" "fmt" "time" ) func cancelByContext(ctx context.Context) { for { select { case <- ctx.Done(): fmt.Println("cancel goroutine by context!") return default: fmt.Println("I'm alive") time.Sleep(1 * time.Second) } } } func main() { ctx, cancel := context.WithCancel(context.Background()) go cancelByContext(ctx) time.Sleep(10*time.Second) cancel() time.Sleep(5*time.Second) }
上面的case中,通過(guò)context自帶的WithCancel方法將cancel函數(shù)傳遞出來(lái),然后手動(dòng)調(diào)用cancel()函數(shù)給goroutine傳遞了ctx.Done()信號(hào)。context也提供了context.WithTimeout()和context.WithDeadline()方法來(lái)更方便的傳遞特定情況下的Done信號(hào)。
package main import ( "context" "fmt" "time" ) func cancelByContext(ctx context.Context) { for { select { case <- ctx.Done(): fmt.Println("cancel goroutine by context!") return default: fmt.Println("I'm alive") time.Sleep(1 * time.Second) } } } func main() { ctx, cancel := context.WithTimeout(context.Background(), time.Second*10) go cancelByContext(ctx) time.Sleep(15*time.Second) }
上述case中使用了context.WithTimeout()來(lái)設(shè)置10秒后自動(dòng)退出,使用context.WithDeadline()的功能基本一樣。區(qū)別是context.WithDeadline()可以指定一個(gè)固定的時(shí)間點(diǎn),當(dāng)然也可以使用time.Now().Add(time.Second*10)的方式來(lái)實(shí)現(xiàn)同context.WithTimeout()相同的功能。具體示例如下:
package main import ( "context" "fmt" "time" ) func cancelByContext(ctx context.Context) { for { select { case <- ctx.Done(): fmt.Println("cancel goroutine by context!") return default: fmt.Println("I'm alive") time.Sleep(1 * time.Second) } } } func main() { ctx, _ := context.WithDeadline(context.Background(), time.Now().Add(time.Second*10)) go cancelByContext(ctx) time.Sleep(15*time.Second) }
注:這里需要注意的一點(diǎn)是上方兩個(gè)case中為了方便讀者理解,我將context傳回的cancel()函數(shù)拋棄掉了,實(shí)際使用中通常會(huì)加上defer cancel()來(lái)保證goroutine被殺死。
附:Context 使用原則和技巧
不要把Context放在結(jié)構(gòu)體中,要以參數(shù)的方式傳遞,parent Context一般為Background
應(yīng)該要把Context作為第一個(gè)參數(shù)傳遞給入口請(qǐng)求和出口請(qǐng)求鏈路上的每一個(gè)函數(shù),放在第一位,變量名建議都統(tǒng)一,如ctx。
給一個(gè)函數(shù)方法傳遞Context的時(shí)候,不要傳遞nil,否則在tarce追蹤的時(shí)候,就會(huì)斷了連接
Context的Value相關(guān)方法應(yīng)該傳遞必須的數(shù)據(jù),不要什么數(shù)據(jù)都使用這個(gè)傳遞
Context是線程安全的,可以放心的在多個(gè)goroutine中傳遞
可以把一個(gè) Context 對(duì)象傳遞給任意個(gè)數(shù)的 gorotuine,對(duì)它執(zhí)行 取消 操作時(shí),所有 goroutine 都會(huì)接收到取消信號(hào)。
這是一種不推薦使用的方法!?。≡诖私o出只是提出這種操作的可能性。實(shí)際場(chǎng)景中尤其是生產(chǎn)環(huán)境請(qǐng)慎用!!
package main import ( "context" "fmt" "time" ) func cancelByPanic(ctx context.Context) { defer func() { if err := recover(); err != nil { fmt.Println("cancel goroutine by panic!") } }() for i:=0 ; i< 5 ;i++{ fmt.Println("hello cancelByPanic") time.Sleep(1 * time.Second) } panic("panic") } func main() { ctx, cancel := context.WithTimeout(context.Background(), time.Second*10) defer cancel() go cancelByPanic(ctx) time.Sleep(5*time.Second) }
這里我們通過(guò)在defer函數(shù)中使用recover來(lái)捕獲panic error并從panic中拿回控制權(quán),確保程序不會(huì)再panic展開到goroutine調(diào)用棧頂部后崩潰。
這是goroutine最常見的退出方式。我們通常都會(huì)等待goroutine執(zhí)行完指定的任務(wù)之后自己退出。所以此處就不給示例了。
了解到goroutine的退出方式后,我們已經(jīng)可以解決一類問(wèn)題。那就是當(dāng)你需要手動(dòng)控制某個(gè)goroutine結(jié)束的時(shí)候應(yīng)該怎么辦。但是在實(shí)際生產(chǎn)中關(guān)于goroutine還有一類問(wèn)題需要解決,那就是當(dāng)你的主進(jìn)程結(jié)束時(shí),應(yīng)該如何等待goroutine全部執(zhí)行完畢后再使主進(jìn)程退出。
阻止程序退出的方法一種有兩種:
package main import ( "fmt" ) func main() { arr := [3]string{"a", "b", "c"} for _, v := range arr { go func(s string) { fmt.Println(s) }(v) } fmt.Println("End") }
以上方的case為例,可見我們?cè)谑裁炊疾患拥臅r(shí)候,不會(huì)等待go func執(zhí)行完主程序就會(huì)退出。因此下面給出使用WaitGroup的方法。
package main import ( "fmt" "sync" ) func main() { var wg sync.WaitGroup // 定義 WaitGroup arr := [3]string{"a", "b", "c"} for _, v := range arr { wg.Add(1) // 增加一個(gè) wait 任務(wù) go func(s string) { defer wg.Done() // 函數(shù)結(jié)束時(shí),通知此 wait 任務(wù)已經(jīng)完成 fmt.Println(s) }(v) } // 等待所有任務(wù)完成 wg.Wait() }
WaitGroup可以理解為一個(gè)goroutine管理者。他需要知道有多少個(gè)goroutine在給他干活,并且在干完的時(shí)候需要通知他干完了,否則他就會(huì)一直等,直到所有的小弟的活都干完為止。我們加上WaitGroup之后,程序會(huì)進(jìn)行等待,直到它收到足夠數(shù)量的Done()信號(hào)為止。
WaitGroup可被調(diào)用的方法只有三個(gè):Add() 、Done()、Wait()。通過(guò)這三個(gè)方法即可實(shí)現(xiàn)上述的功能,下面我們把源碼貼出。
func (wg *WaitGroup) Add(delta int) { statep := wg.state() state := atomic.AddUint64(statep, uint64(delta)<<32) v := int32(state >> 32) // 計(jì)數(shù)器 w := uint32(state) // 等待者個(gè)數(shù)。這里用uint32,會(huì)直接截?cái)嗔烁呶?2位,留下低32位 if v < 0 { // Done的執(zhí)行次數(shù)超出Add的數(shù)量 panic("sync: negative WaitGroup counter") } if w != 0 && delta > 0 && v == int32(delta) { // 最開始時(shí),Wait不能在Add之前被執(zhí)行 panic("sync: WaitGroup misuse: Add called concurrently with Wait") } if v > 0 || w == 0 { // 計(jì)數(shù)器不為零,還有沒Done的。return // 沒有等待者。return return } // 所有g(shù)oroutine都完成任務(wù)了,但有g(shù)oroutine執(zhí)行了Wait后被阻塞,需要喚醒它 if *statep != state { // 已經(jīng)到了喚醒階段了,就不能同時(shí)并發(fā)Add了 panic("sync: WaitGroup misuse: Add called concurrently with Wait") } // 清零之后,就可以繼續(xù)Add和Done了 *statep = 0 for ; w != 0; w-- { // 喚醒 runtime_Semrelease(&wg.sema, false) } } func (wg *WaitGroup) Done() { wg.Add(-1) } func (wg *WaitGroup) Wait() { statep := wg.state() for { state := atomic.LoadUint64(statep) v := int32(state >> 32) // 計(jì)數(shù)器 w := uint32(state) // 等待者個(gè)數(shù) if v == 0 { // 如果聲明變量后,直接執(zhí)行Wait也不會(huì)有問(wèn)題 // 下面CAS操作失敗,重試,但剛好發(fā)現(xiàn)計(jì)數(shù)器變成零了,安全退出 return } if atomic.CompareAndSwapUint64(statep, state, state+1) { if race.Enabled && w == 0 { race.Write(unsafe.Pointer(&wg.sema)) } // 掛起當(dāng)前的g runtime_Semacquire(&wg.sema) // 被喚醒后,計(jì)數(shù)器不應(yīng)該大于0 // 大于0意味著Add的數(shù)量被Done完后,又開始了新一波Add if *statep != 0 { panic("sync: WaitGroup is reused before previous Wait has returned") } return } } }
通過(guò)看源碼,我們可以知道,有些使用細(xì)節(jié)是需要注意的:
1.wg.Done()函數(shù)實(shí)際上實(shí)現(xiàn)的是wg.Add(-1),因此直接使用wg.Add(-1)是會(huì)造成同樣的結(jié)果的。在實(shí)際使用中要注意避免誤操作,使得監(jiān)聽的goroutine數(shù)量出現(xiàn)誤差。
2.wg.Add()函數(shù)可以一次性加n。但是實(shí)際使用時(shí)通常都設(shè)為1。但是wg本身的counter不能設(shè)為負(fù)數(shù)。假設(shè)你在沒有Add到10以前,一次性wg.Add(-10),會(huì)出現(xiàn)panic !
package main import ( "fmt" "sync" ) func main() { var wg sync.WaitGroup // 定義 WaitGroup arr := [3]string{"a", "b", "c"} for _, v := range arr { wg.Add(1) // 增加一個(gè) wait 任務(wù) go func(s string) { defer wg.Done() // 函數(shù)結(jié)束時(shí),通知此 wait 任務(wù)已經(jīng)完成 fmt.Println(s) }(v) } wg.Add(-10) // 等待所有任務(wù)完成 wg.Wait() } panic: sync: negative WaitGroup counter
3.如果你的程序?qū)懙挠袉?wèn)題,出現(xiàn)了始終等待的waitgroup會(huì)造成死鎖。
package main import ( "fmt" "sync" ) func main() { var wg sync.WaitGroup // 定義 WaitGroup arr := [3]string{"a", "b", "c"} for _, v := range arr { wg.Add(1) // 增加一個(gè) wait 任務(wù) go func(s string) { defer wg.Done() // 函數(shù)結(jié)束時(shí),通知此 wait 任務(wù)已經(jīng)完成 fmt.Println(s) }(v) } wg.Add(1) // 等待所有任務(wù)完成 wg.Wait() } fatal error: all goroutines are asleep - deadlock!
第二種方法即是通過(guò)channel。具體寫法如下:
package main import "fmt" func main() { arr := [3]string{"a", "b", "c"} ch := make(chan struct{}, len(arr)) for _, v := range arr { go func(s string) { fmt.Println(s) ch <- struct{}{} }(v) } for i := 0; i < len(arr); i ++ { <-ch } }
需要注意的是,channel同樣會(huì)導(dǎo)致死鎖。如下方示例:
package main import "fmt" func main() { arr := [3]string{"a", "b", "c"} ch := make(chan struct{}, len(arr)) for _, v := range arr { go func(s string) { fmt.Println(s) ch <- struct{}{} }(v) } for i := 0; i < len(arr); i++ { <-ch } <-ch } fatal error: all goroutines are asleep - deadlock!
利用go routine的這一特性,我們可以將waitGroup等方式封裝起來(lái),保證go routine在主進(jìn)程結(jié)束時(shí)會(huì)繼續(xù)執(zhí)行完。封裝demo:
package main import ( "fmt" "sync" ) type WaitGroupWrapper struct { sync.WaitGroup } func (wg *WaitGroupWrapper) Wrap(f func(args ...interface{}), args ...interface{}) { wg.Add(1) go func() { f(args...) wg.Done() }() } func printArray(args ...interface{}){ fmt.Println(args) } func main() { var w WaitGroupWrapper // 定義 WaitGroup arr := [3]string{"a", "b", "c"} for _, v := range arr { w.Wrap(printArray,v) } w.Wait() }
還可以加上更高端一點(diǎn)的功能,增加時(shí)間、事件雙控制的wrapper。
package main import ( "fmt" "sync" "time" ) type WaitGroupWrapper struct { sync.WaitGroup } func (wg *WaitGroupWrapper) Wrap(f func(args ...interface{}), args ...interface{}) { wg.Add(1) go func() { f(args...) wg.Done() }() } func (w *WaitGroupWrapper) WaitWithTimeout(d time.Duration) bool { ch := make(chan struct{}) t := time.NewTimer(d) defer t.Stop() go func() { w.Wait() ch <- struct{}{} }() select { case <-ch: fmt.Println("job is done!") return true case <-t.C: fmt.Println("time is out!") return false } } func printArray(args ...interface{}){ time.Sleep(3*time.Second) //3秒后會(huì)觸發(fā)time is out分支 //如果改為time.Sleep(time.Second)即會(huì)觸發(fā)job is done分支 fmt.Println(args) } func main() { var w WaitGroupWrapper // 定義 WaitGroup arr := [3]string{"a", "b", "c"} for _, v := range arr { w.Wrap(printArray,v) } w.WaitWithTimeout(2*time.Second) }
到此,相信大家對(duì)“Go語(yǔ)言中Goroutine退出機(jī)制如何使用”有了更深的了解,不妨來(lái)實(shí)際操作一番吧!這里是億速云網(wǎng)站,更多相關(guān)內(nèi)容可以進(jìn)入相關(guān)頻道進(jìn)行查詢,關(guān)注我們,繼續(xù)學(xué)習(xí)!
免責(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)容。