前言 在golang中,只需要在函數調用前加上關鍵字go即可創建一個併發任務單元,而這個新建的任務會被放入隊列中,等待調度器安排。相比系統的MB級別線程棧,goroutine的自定義棧只有2KB,這使得我們能夠輕易創建上萬個併發任務,如此對性能提升不少。但隨之而來的有以下幾個問題: "如何等待所有g ...
前言
在golang中,只需要在函數調用前加上關鍵字go即可創建一個併發任務單元,而這個新建的任務會被放入隊列中,等待調度器安排。相比系統的MB級別線程棧,goroutine的自定義棧只有2KB,這使得我們能夠輕易創建上萬個併發任務,如此對性能提升不少。但隨之而來的有以下幾個問題:
本文記錄了筆者就以上幾個問題進行探究的過程,文中給出了大部分問題的解決方案,同時也拋出了未解決的問題,期待與各位交流:p
準備
開始之前先定義一個常量const N=100
以及一個HeavyWork
函數,假定該函數具有極其冗長、複雜度高、難以解耦的特性
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)
}
以上定義的內容將在之後的代碼中直接使用以縮減篇幅,大部分完整代碼可在 Github: explore-goroutine 中找到
如何等待所有goroutine的退出
"Do not communicate by sharing memory; instead, share memory by communicating"——GO的一大設計哲學《Share Memory By Communicating》
翻譯成中文就是,用通信來共用記憶體數據,而不要通過共用記憶體數據來進行通信。
Go中的goroutines和channel提供了一種優雅而獨特的結構化併發軟體的方法,我們可以利用通道(channel)的特性,來實現當前等待goroutine的操作。但是channel並不是當前這個場景的最佳方案,用它來實現的方式是稍顯笨拙的,需要知道確定個數的goroutine,同時稍不註意就極易產生死鎖,代碼如下:
// "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")
}
上述代碼使用了一個緩存大小為1的通道(channel),創建N個goroutine用於運行HeavyWork
,每個任務完成後向waitChan寫入一個數據,在收到N個完成信號後退出。
但事實上比較優雅的方式是使用go標準庫sync
,其中提供了專門的解決方案sync.WaitGroup
用於等待一個goroutines集合的結束
// "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")
}
關於sync.WaitGroup
的具體使用請參照官方文檔 [GoDoc] sync.WaitGroup ,這裡不再贅述
如何限制goroutine的創建數量(信號量實現)
信號量(Semaphore),有時被稱為信號燈,是在多線程環境下使用的一種設施,是可以用來保證兩個或多個關鍵代碼段不被併發調用。
其中V操作會增加信號量的數值即釋放資源,而P操作會減少它即占用資源
那麼非常容易想到的就是利用channel(通道)緩存有限的特性,它允許我們可以自實現一個簡單的數量控制,就如同使用信號量一般,在這基礎再加上前面提到的sync.WaitGroup
,我們可以打出一套組合拳,提供可阻塞的信號量PV操作,能夠實現固定創建goroutine數量並且支持等待當前goroutine的退出。結構體定義如下:
type Semaphore struct {
Threads chan int
Wg sync.WaitGroup
}
而P操作只需在channel中加入一個元素同時調用WaitGroup.Add即可,這一操作完成對資源的申請
func (sem *Semaphore) P() {
sem.Threads <- 1
sem.Wg.Add(1)
}
相反則是V操作,進行資源的釋放
func (sem *Semaphore) V() {
sem.Wg.Done()
<-sem.Threads
}
Wait則阻塞等待直到當前所有資源都歸還,直接調用WaitGroup的方法即可
func (sem *Semaphore) Wait() {
sem.Wg.Wait()
}
完整代碼可以在 Github: semaphore 中查看
利用上面的信號量就可以做到,在一個時刻的goroutines數量不會超過信號量值的大小,而某個goroutine退出後將返還占用的信號量,而正在等待的goroutine就可以立即申請,下圖形象地展現了運行時的狀態
怎麼讓goroutine主動退出
對於goroutine的主動退出,比較友好的做法就是迴圈監聽一個channel,通過類似信號的方式來告知goroutine的”該退出了“,然後goroutine自己主動退出,這種做法在網上十分常見,也是Golang官方推薦的做法,思想也很簡單。
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
}
運行結果如下圖
探索——如何從外部殺死goroutine
上面講了一些關於goroutines和channel的簡單使用,接下來終於寫到本文的重點了。筆者並沒有解決如何從外部殺死一個goroutine,但記錄了嘗試“殺死”中的可行或不可行方法,希望對各位有所幫助。
因為近期在開發中遇到這樣一個問題,當一個函數是極其冗長、複雜度高、難以解耦的順序結構代碼時(例如某個極其複雜無迴圈結構的加密演算法),而且由於數據量巨大,需要反覆調用該函數,由於每運行一次,程式都會消耗大量的時間、空間,那麼當一個任務已經被用戶拋棄時,如何才能拋棄仍在做著無用功的goroutine?
為了達到“殺死goroutine”的目的,筆者做了很多嘗試,如
- select結構(條件實現)
- panic退出機制(失敗)
- 獲取pid殺死(失敗)
- ptrace單步調試(失敗)
- ...(失敗)
利用select語句實現
關於“如何殺死goroutine”,網上有一部分答案就是利用select實現的,但是這種方式實現的代碼並不適用於服務類的程式,但是對於一般非服務類程式的確能夠實現殺死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)
}
但是一旦主函數沒有立即退出,而是作為某種服務而繼續運行時,這裡刪除了main函數的最後一行註釋time.Sleep(3 * time.Second)
,延遲三秒後退出。可以看見儘管已經超時並輸出"time limit exceed"之後,HeavyWork
在main函數沒退出前依舊在運行。效果如下
所以使用select-timeout的方式比較適合實時退出類型的程式,能夠實現一定程度上的併發控制,
小結
就目前而言,還沒有完美的方案來解決控制goroutine的問題,事實上Go似乎並不允許和推薦人們直接控制goroutine,所以暫時還無法做到從外部直接控制goroutine的生命周期,所以比較推薦的做法還是只能通過goroutine主動退出的方法,迴圈監聽channel,在發出退出信號後最多只消耗一輪資源後就退出,但這就要求該代碼具有迴圈結構否則就很難使用。有更好解決方案的朋友,請務必告訴我!
轉載請註明出處:http://www.cnblogs.com/tr3e/p/7995689.html