前言

在golang中，只需要在函數(shù)調(diào)用前加上關(guān)鍵字go即可創(chuàng)建一個(gè)并發(fā)任務(wù)單元，而這個(gè)新建的任務(wù)會(huì)被放入隊(duì)列中，等待調(diào)度器安排。相比系統(tǒng)的MB級(jí)別線(xiàn)程棧，goroutine的自定義棧只有2KB，這使得我們能夠輕易創(chuàng)建上萬(wàn)個(gè)并發(fā)任務(wù)，如此對(duì)性能提升不少。但隨之而來(lái)的有以下幾個(gè)問(wèn)題：

如何等待所有g(shù)oroutine的退出

如何限制創(chuàng)建goroutine的數(shù)量（信號(hào)量實(shí)現(xiàn)）

怎么讓goroutine主動(dòng)退出

探索——如何從外部殺死goroutine

本文記錄了筆者就以上幾個(gè)問(wèn)題進(jìn)行探究的過(guò)程，文中給出了大部分問(wèn)題的解決方案，同時(shí)也拋出了未解決的問(wèn)題，期待與各位交流：p

準(zhǔn)備

開(kāi)始之前先定義一個(gè)常量const N=100以及一個(gè)HeavyWork函數(shù)，假定該函數(shù)具有極其冗長(zhǎng)、復(fù)雜度高、難以解耦的特性

func HeavyWork(id int) {
 rand.Seed(int64(id))
 interval := time.Duration(rand.Intn(3)+1) * time.Second
 time.Sleep(interval)
 fmt.Printf("HeavyWork %-3d cost %v\n", id, interval)
}

以上定義的內(nèi)容將在之后的代碼中直接使用以縮減篇幅，大部分完整代碼可在 Github: explore-goroutine 中找到

如何等待所有g(shù)oroutine的退出

"Do not communicate by sharing memory; instead, share memory by communicating"——GO的一大設(shè)計(jì)哲學(xué)《Share Memory By Communicating》

翻譯成中文就是，用通信來(lái)共享內(nèi)存數(shù)據(jù)，而不要通過(guò)共享內(nèi)存數(shù)據(jù)來(lái)進(jìn)行通信。

Go中的goroutines和channel提供了一種優(yōu)雅而獨(dú)特的結(jié)構(gòu)化并發(fā)軟件的方法，我們可以利用通道(channel)的特性，來(lái)實(shí)現(xiàn)當(dāng)前等待goroutine的操作。但是channel并不是當(dāng)前這個(gè)場(chǎng)景的最佳方案，用它來(lái)實(shí)現(xiàn)的方式是稍顯笨拙的，需要知道確定個(gè)數(shù)的goroutine，同時(shí)稍不注意就極易產(chǎn)生死鎖，代碼如下：

// "talk is cheap, show me the code."
func main() {
 waitChan := make(chan int, 1)
 for i := 0; i < N; i++ {
  go func(n int) {
   HeavyWork(n)
   waitChan <- 1
  }(i)
 }
 cnt := 0
 for range waitChan {
  cnt++
  if cnt == N {
   break
  }
 }
 close(waitChan)
 fmt.Println("finished")
}

上述代碼使用了一個(gè)緩存大小為1的通道(channel)，創(chuàng)建N個(gè)goroutine用于運(yùn)行HeavyWork，每個(gè)任務(wù)完成后向waitChan寫(xiě)入一個(gè)數(shù)據(jù)，在收到N個(gè)完成信號(hào)后退出。

但事實(shí)上比較優(yōu)雅的方式是使用go標(biāo)準(zhǔn)庫(kù)sync，其中提供了專(zhuān)門(mén)的解決方案sync.WaitGroup用于等待一個(gè)goroutines集合的結(jié)束

// "talk is cheap, show me the code."
func main() {
 wg := sync.WaitGroup{}
 for i := 0; i < N; i++ {
  wg.Add(1)
  go func(n int) {
   defer wg.Done()
   HeavyWork(n)
  }(i)
 }
 wg.Wait()
 fmt.Println("finished")
}

關(guān)于sync.WaitGroup的具體使用請(qǐng)參照官方文檔[GoDoc] sync.WaitGroup ，這里不再贅述

如何限制goroutine的創(chuàng)建數(shù)量（信號(hào)量實(shí)現(xiàn)）

信號(hào)量(Semaphore)，有時(shí)被稱(chēng)為信號(hào)燈，是在多線(xiàn)程環(huán)境下使用的一種設(shè)施，是可以用來(lái)保證兩個(gè)或多個(gè)關(guān)鍵代碼段不被并發(fā)調(diào)用。

其中V操作會(huì)增加信號(hào)量的數(shù)值即釋放資源，而P操作會(huì)減少它即占用資源

那么非常容易想到的就是利用channel（通道）緩存有限的特性，它允許我們可以自實(shí)現(xiàn)一個(gè)簡(jiǎn)單的數(shù)量控制，就如同使用信號(hào)量一般，在這基礎(chǔ)再加上前面提到的sync.WaitGroup，我們可以打出一套組合拳，提供可阻塞的信號(hào)量PV操作，能夠?qū)崿F(xiàn)固定創(chuàng)建goroutine數(shù)量并且支持等待當(dāng)前goroutine的退出。結(jié)構(gòu)體定義如下：

type Semaphore struct {
 Threads chan int
 Wg  sync.WaitGroup
}

而P操作只需在channel中加入一個(gè)元素同時(shí)調(diào)用WaitGroup.Add即可，這一操作完成對(duì)資源的申請(qǐng)

func (sem *Semaphore) P() {
 sem.Threads <- 1
 sem.Wg.Add(1)
}

相反則是V操作，進(jìn)行資源的釋放

func (sem *Semaphore) V() {
 sem.Wg.Done()
 <-sem.Threads
}

Wait則阻塞等待直到當(dāng)前所有資源都?xì)w還，直接調(diào)用WaitGroup的方法即可

func (sem *Semaphore) Wait() {
 sem.Wg.Wait()
}

完整代碼可以在 Github: semaphore 中查看

利用上面的信號(hào)量就可以做到，在一個(gè)時(shí)刻的goroutines數(shù)量不會(huì)超過(guò)信號(hào)量值的大小，而某個(gè)goroutine退出后將返還占用的信號(hào)量，而正在等待的goroutine就可以立即申請(qǐng)，下圖形象地展現(xiàn)了運(yùn)行時(shí)的狀態(tài)

怎么讓goroutine主動(dòng)退出

對(duì)于goroutine的主動(dòng)退出，比較友好的做法就是循環(huán)監(jiān)聽(tīng)一個(gè)channel，通過(guò)類(lèi)似信號(hào)的方式來(lái)告知goroutine的”該退出了“，然后goroutine自己主動(dòng)退出，這種做法在網(wǎng)上十分常見(jiàn)，也是Golang官方推薦的做法，思想也很簡(jiǎn)單。

func main() {
 ok, quit := make(chan int, 1), make(chan int, 1)
 go func() {
  i := 0
  for {
   select {
   case <-quit:
    ok <- 1
    return
   default:
    HeavyWork(i)
    i++
   }
  }
 }()
 time.Sleep(5 * time.Second)
 quit <- 1
 <-ok
}

運(yùn)行結(jié)果如下圖

探索——如何從外部殺死goroutine

上面講了一些關(guān)于goroutines和channel的簡(jiǎn)單使用，接下來(lái)終于寫(xiě)到本文的重點(diǎn)了。筆者并沒(méi)有解決如何從外部殺死一個(gè)goroutine，但記錄了嘗試“殺死”中的可行或不可行方法，希望對(duì)各位有所幫助。

因?yàn)榻谠陂_(kāi)發(fā)中遇到這樣一個(gè)問(wèn)題，當(dāng)一個(gè)函數(shù)是極其冗長(zhǎng)、復(fù)雜度高、難以解耦的順序結(jié)構(gòu)代碼時(shí)（例如某個(gè)極其復(fù)雜無(wú)循環(huán)結(jié)構(gòu)的加密算法），而且由于數(shù)據(jù)量巨大，需要反復(fù)調(diào)用該函數(shù)，由于每運(yùn)行一次，程序都會(huì)消耗大量的時(shí)間、空間，那么當(dāng)一個(gè)任務(wù)已經(jīng)被用戶(hù)拋棄時(shí)，如何才能拋棄仍在做著無(wú)用功的goroutine？

為了達(dá)到“殺死goroutine”的目的，筆者做了很多嘗試，如

select結(jié)構(gòu)（條件實(shí)現(xiàn)）

panic退出機(jī)制（失?。?/p>

獲取pid殺死（失?。?/p>

ptrace單步調(diào)試（失?。?/p>

...（失?。?/p>

利用select語(yǔ)句實(shí)現(xiàn)

關(guān)于“如何殺死goroutine”，網(wǎng)上有一部分答案就是利用select實(shí)現(xiàn)的，但是這種方式實(shí)現(xiàn)的代碼并不適用于服務(wù)類(lèi)的程序，但是對(duì)于一般非服務(wù)類(lèi)程序的確能夠?qū)崿F(xiàn)殺死goroutine的效果，代碼如下：

func main() {
 wrapper := func() chan int {
  c := make(chan int)
  go func() {
   HeavyWork(0)
   c <- 1
  }()
  return c
 }
 select {
 case <-wrapper():
 case <-time.After(1 * time.Second):
  fmt.Println("time limit exceed")
 }
 // time.Sleep(3 * time.Second)
}

但是一旦主函數(shù)沒(méi)有立即退出，而是作為某種服務(wù)而繼續(xù)運(yùn)行時(shí)，這里刪除了main函數(shù)的最后一行注釋time.Sleep(3 * time.Second)，延遲三秒后退出?？梢钥匆?jiàn)盡管已經(jīng)超時(shí)并輸出"time limit exceed"之后，HeavyWork在main函數(shù)沒(méi)退出前依舊在運(yùn)行。效果如下

所以使用select-timeout的方式比較適合實(shí)時(shí)退出類(lèi)型的程序，能夠?qū)崿F(xiàn)一定程度上的并發(fā)控制，

小結(jié)

就目前而言，還沒(méi)有完美的方案來(lái)解決控制goroutine的問(wèn)題，事實(shí)上Go似乎并不允許和推薦人們直接控制goroutine，所以暫時(shí)還無(wú)法做到從外部直接控制goroutine的生命周期，所以比較推薦的做法還是只能通過(guò)goroutine主動(dòng)退出的方法，循環(huán)監(jiān)聽(tīng)channel，在發(fā)出退出信號(hào)后最多只消耗一輪資源后就退出，但這就要求該代碼具有循環(huán)結(jié)構(gòu)否則就很難使用。有更好解決方案的朋友，請(qǐng)務(wù)必告訴我！

以上這篇Golang 探索對(duì)Goroutine的控制方法(詳解)就是小編分享給大家的全部?jī)?nèi)容了，希望能給大家一個(gè)參考，也希望大家多多支持腳本之家。

最新評(píng)論