聊聊Golang的協程洩露，看看怎麼預防洩露
協程在Go裡面是一個常見的概念，伴隨著Go程式的生命週期開始至結束。今天來細聊Go的協程洩露。【相關推薦：Go視訊教學】
洩露案例
關於協程洩露很多時候我們往往會忽略它，直到機器資源負載異常才引起重視。 之前排除生產環境異常的時候，曾經遇到過go程式記憶體洩露的場景，記憶體漏失和協程洩露有很大關係，本質上都是資源不回收導致的。
這裡列舉一個典型的洩露案例：
func JumpForSignal() int {
ch := make(chan int)
go func() {
ch <- bizMtx
}()
go func() {
ch <- bizMtx
}()
go func() {
ch <- bizMtx
}()
//一有輸入立刻返回
return <-ch
}
func main() {
// ...
JumpForSignal()
// ...
}事後分析這個demo可以得知，這個函數呼叫會阻塞兩個子協程，預期只有一個協程會正常退出。
獲取協程資訊
既然存在協程洩露，我們在日常工作怎麼避免或者發現它呢？下面我們列舉幾個思路。
遵守準則
由於Go是自帶GC的語言，很多時候寫程式碼不需要關心變數的資源釋放，不像C程式設計師變數申請之後需要在結束處釋放。但是Go的chan在使用時候是有一些準則的，當確定chan不再使用時候，可以在輸出方進行close，避免其他協程還在等待該chan的輸出。
協程數量
找到洩露的協程，第一個能夠想到的是協程數量，當你的函數處理邏輯比較簡單，除了主協程之外，預期協程應該都在結束前返回，可以在main函數結束處呼叫runtime包的函數：
// NumGoroutine returns the number of goroutines that currently exist.
func NumGoroutine() int {
return int(gcount())
}通過它可以返回當前協程總數量：
func Count() {
fmt.Printf("Number of goroutines:%d\n", runtime.NumGoroutine())
}
func main() {
defer Count()
Count()
JumpForSignal()
}輸出：
Number of goroutines:1 Number of goroutines:3
協程函數棧
還有一種比較常見定位協程的形式，在Go裡面，可以用於分析協程函數的上下文，常見的比如go自帶的pprof也是通過這種方式獲取，實際案例中，條件允許的情況可以開啟pprof方便分析。
下面來看一個範例，我們在上面的例子加一個http埠監聽，用於接入go自帶的pprof分析工具。
隨後在瀏覽器輸入：
http://localhost:8899/debug/pprof/goroutine?debug=1
可以得到整個程式的協程列表：
goroutine profile: total 7 1 @ 0x165eb6 0x126465 0x126235 0x29341e 0x19de01 # 0x29341d pixelgo/leak.JumpForSignal.func1+0x3d F:/code/pixelGo/src/pix-demo/leak/leak.go:24 1 @ 0x165eb6 0x126465 0x126235 0x29347e 0x19de01 # 0x29347d pixelgo/leak.JumpForSignal.func2+0x3d F:/code/pixelGo/src/pix-demo/leak/leak.go:28 1 @ 0x165eb6 0x15bb3d 0x1975a5 0x228d05 0x229d8d 0x22c40d 0x321765 0x33437c 0x447c89 0x285239 0x285606 0x4493f3 0x450da8 0x19de01 # 0x1975a4 internal/poll.runtime_pollWait+0x64 D:/dev/go1.16/src/runtime/netpoll.go:227 # 0x228d04 internal/poll.(*pollDesc).wait+0xa4 D:/dev/go1.16/src/internal/poll/fd_poll_runtime.go:87 # 0x229d8c internal/poll.execIO+0x2ac D:/dev/go1.16/src/internal/poll/fd_windows.go:175 # 0x22c40c internal/poll.(*FD).Read+0x56c // ...
結論是：當前程式一共有7個協程，可以看出分別有1個協程分配在F:/code/pixelGo/src/pix-demo/leak/leak.go:24 和F:/code/pixelGo/src/pix-demo/leak/leak.go:28，正是上文洩露的程式碼塊。
有時候還可以多維度去分析，比如輸入：
http://localhost:8899/debug/pprof/goroutine?debug=2
可以通過協程後面的標籤,看到當前協程的不同狀態，running/io wait/chan send
goroutine 9 [running]:
runtime/pprof.writeGoroutineStacks(0x7f7d00, 0xc0000aa000, 0x0, 0x0)
D:/dev/go1.16/src/runtime/pprof/pprof.go:693 +0xc5
net/http/pprof.handler.ServeHTTP(0xc000094011, 0x9, 0x7fba40, 0xc0000aa000, 0xc000092000)
//..
goroutine 1 [IO wait]:
internal/poll.runtime_pollWait(0x223debb10d8, 0x72, 0xc000152f48)
D:/dev/go1.16/src/runtime/netpoll.go:227 +0x65
internal/poll.(*pollDesc).wait(0xc0001530b8, 0x72, 0x93b400, 0x0, 0x0)
//...
goroutine 6 [chan send]:
pixelgo/rout.JumpForSignal.func1(0xc000053800)
F:/code/pixelGo/src/pix-demo/rout/leak.go:25 +0x10e
created by pixelgo/rout.JumpForSignal
F:/code/pixelGo/src/pix-demo/rout/leak.go:23 +0x71
goroutine 7 [chan send]:
pixelgo/rout.JumpForSignal.func2(0xc000053800)
F:/code/pixelGo/src/pix-demo/rout/leak.go:30 +0x10e
created by pixelgo/rout.JumpForSignal
F:/code/pixelGo/src/pix-demo/rout/leak.go:28 +0x93協程id
接下來我們來探索協程標識：協程id，在Go中，每個執行的協程都會分配一個協程id，一個常見的方式是從函數執行棧獲取，參照之前網上其他同學的寫法：
func main() {
fmt.Println(getGID())
}
func getGID() uint64 {
b := make([]byte, 64)
b = b[:runtime.Stack(b, false)]
b = bytes.TrimPrefix(b, []byte("goroutine "))
b = b[:bytes.IndexByte(b, ' ')]
n, _ := strconv.ParseUint(string(b), 10, 64)
return n
}我們來看看runtime.stack() 會返回什麼呢，其中真實內容是這樣的：
goroutine 21 [running]: leaktest.interestingGoroutines(0xdb9980, 0xc00038e018, 0x0, 0x0, 0x0) F:/code/pixelGo/src/leaktest/leaktest.go:81 +0xbf leaktest.CheckContext(0xdbe398, 0xc000108040, 0xdb9980, 0xc00038e018, 0x0) F:/code/pixelGo/src/leaktest/leaktest.go:141 +0x6e leaktest.CheckTimeout(0xdb9980, 0xc00038e018, 0x3b9aca00, 0x0) F:/code/pixelGo/src/leaktest/leaktest.go:127 +0xe5 leaktest.TestCheck.func8(0xc000384780) F:/code/pixelGo/src/leaktest/leaktest_test.go:122 +0xaf testing.tRunner(0xc000384780, 0xc000100050) D:/dev/go1.16/src/testing/testing.go:1193 +0x1a3 created by testing.(*T).Run D:/dev/go1.16/src/testing/testing.go:1238 +0x63c goroutine 1 [chan receive]: testing.(*T).Run(0xc000037080, 0xd8486a, 0x9, 0xd9ebc8, 0x304bd824304bd800) D:/dev/go1.16/src/testing/testing.go:1239 +0x66a testing.runTests.func1(0xc000036f00) D:/dev/go1.16/src/testing/testing.go:1511 +0xbd testing.tRunner(0xc000036f00, 0xc00008fc00) D:/dev/go1.16/src/testing/testing.go:1193 +0x1a3 testing.runTests(0xc0000040d8, 0xf40460, 0x5, 0x5, 0x0, 0x0, 0x0, 0x21cbf1c0100) D:/dev/go1.16/src/testing/testing.go:1509 +0x448 testing.(*M).Run(0xc0000c0000, 0x0) D:/dev/go1.16/src/testing/testing.go:1417 +0x514 main.main() _testmain.go:51 +0xc8
可以發現這個棧和我們執行panic丟擲的資訊非常類似，需要注意的是，通過這種方式獲取協程id並不是一個高效的方式。
實際生產使用過程並不提倡，值得一提的是，為了方便我們更好的定位問題上下文，有時候紀錄檔框架又需要我們列印出當前協程id。
比如這是一個生產案例紀錄檔輸出：
// gid-1號協程用於初始化資源
[0224/162532.310:INFO:gid-1:yx_trace.go:66] cfg:&{ false false [] 0xc000295140 0xc0001d4e00 <nil> <nil> <nil>}
[0224/162532.320:INFO:gid-1:main.go:50] GameRoom Startup->
[0224/162532.320:INFO:gid-1:config_manager.go:107] configManager SetHttpListenAddr:8080
[0224/162532.320:INFO:gid-1:room_manager.go:57] roomManager Startup
[0224/162532.323:INFO:gid-1:room_manager.go:72] roomManager initPrx.
[0224/162532.330:INFO:gid-1:bootstrap.go:153] GameRoom START ok.
// gid-60號協程分配用於啟動HTTP Server
[0224/162533.277:INFO:gid-60:expose.go:36] Start for HTTP server...
[0224/162533.277:INFO:gid-60:expose.go:39] register for debug server...往往紀錄檔框架是力求對業務效能影響最低的，既然有效能顧慮，那麼它是怎麼獲取協程id的呢？只能曲線救國了。
還有一個解法，其實在Go中，每個協程繫結的系統執行緒結構中，有一個g指標，拿到g指標的資訊之後，根據g指標結構的偏移量(注意不同go版本可能不同)，指定獲取id。
組合獲取
通過協程繫結的g指標，這裡參考《Go高階程式設計》的做法
// 記錄各個版本的偏移量
var offsetDictMap = map[string]int64{
"go1.12": 152,
"go1.12.1": 152,
"go1.12.2": 152,
"go1.12.3": 152,
"go1.12.4": 152,
"go1.12.5": 152,
"go1.12.6": 152,
"go1.12.7": 152,
"go1.13": 152,
"go1.14": 152,
"go1.16.12": 152,
}
// offset for go1.12
var goid_offset uintptr = 152
//go:nosplit
func getG() interface{}
func GoId() int64
// 部分組合程式碼
// func getGptr() unsafe.Pointer
TEXT ·getGptr(SB), NOSPLIT, $0-8
MOVQ (TLS), BX
MOVQ BX, ret+0(FP)
RET
TEXT ·GoId(SB),NOSPLIT,$0-8
NO_LOCAL_POINTERS
MOVQ ·goid_offset(SB),AX
// get runtime.g
MOVQ (TLS),BX
ADDQ BX,AX
MOVQ (AX),BX
MOVQ BX,ret+0(FP)
RET這裡點到為止，大概思路是這樣。
效能比較：
我們來簡單測試下兩種獲取go協程id方式效能差距：
// BenchmarkGRtId-8 1000000000 0.0005081 ns/op
func BenchmarkGRtId(b *testing.B) {
for n := 0; n < 1000000000; n++ {
// runtime獲取協程id
getGID()
}
}
// BenchmarkGoId-8 1000000000 0.05731 ns/op
func BenchmarkGoId(b *testing.B) {
for n := 0; n < 1000000000; n++ {
// 組合方式獲取
GoId()
}
}可以看到通過組合方式獲取協程id的方式效能更優，相差幾個數量級。
限制協程
上面列舉了幾個定位協程資訊的方法，那麼在協程洩露之前有沒有其他方式對程式的go協程進行管控呢，有個做法是使用強大的channel坐下限制。
拋磚引玉
這裡先提供一個簡單的思路，即再包裝一層channel進行保護，
// 限制數量
var LIMIT_G_NUM = make(chan struct{}, 100)
// 需要自定義的處理邏輯
type HandleFun func()
func AsyncGoForHandle(fn HandleFun) {
// 計數加一
LIMIT_G_NUM <- struct{}{}
go func() {
defer func() {
if err := recover(); err != nil {
log.Fatalf("AsyncGoForHandle recover from err: %v", err)
}
// 回收計數
<-LIMIT_G_NUM
}()
// 處理邏輯
fn()
}()
}上面的思路比較簡單，相信大家能看懂，每次需要非同步建立協程只要呼叫AsyncGoForHandle()函數即可，不足之處可能是處理邏輯HandleFun()不夠通用，需要自己定義具體實現。
還有一種方式，就是引入協程池的概念，這裡的池子和資料庫連線池有點像，即一開始就預建立好，業務層只要負責提交資料，業界已經有不少成熟的封裝。
成熟方案：tunny
之前看到社群有一個封裝得比較完善的協程池tunny，程式碼行數不多，我們來試著拆解分析一下程式碼，專案地址：https://github.com/Jeffail/tunny
1、定義處理邏輯介面：
type Worker interface {
// 自定義邏輯實現，開發者只需要關心入參和出參
Process(interface{}) interface{}
}2、包裝worker的輸入源workRequest
type workerWrapper struct {
// 注入內部實現邏輯
worker Worker
interruptChan chan struct{}
// 請求來源workRequest
reqChan chan<- workRequest
// ...
}3、輸入源結構
type workRequest struct {
// 輸入
jobChan chan<- interface{}
// 處理結果，即worker.Process()的返回值
retChan <-chan interface{}
// ...
}4、編寫實現類：
我們知道Go的介面遵循鴨子模型： 只要它表現得像個鴨子，它就是鴨子
// Worker實現類
type closureWorker struct {
processor func(interface{}) interface{}
}
func (w *closureWorker) Process(payload interface{}) interface{} {
return w.processor(payload)
}5、定義工作池結構
type Pool struct {
queuedJobs int64
// 成員函數，用於"鴨子"實體
ctor func() Worker
workers []*workerWrapper
reqChan chan workRequest
workerMut sync.Mutex
}
func NewFunc(n int, f func(interface{}) interface{}) *Pool {
return New(n, func() Worker {
return &closureWorker{
// 傳入真正的實現模組
processor: f,
}
})
}
func New(n int, ctor func() Worker) *Pool {
p := &Pool{
ctor: ctor,
reqChan: make(chan workRequest),
}
// 批次建立協程，監聽處理來自reqChan的任務
p.SetSize(n)
return p
}相關實體結構如下，配合原始碼閱讀就比較清晰了。
這個框架相當於把協程預先建立好做了池化，隨後業務層只需要源源不斷把"加工資料"輸入到workRequest這個chan即可，也就是process()函數，process()模組會把資料輸入到內部channel進行處理，池中的worker會進行加工。
這種工廠模式還是值得借鑑的，Go也有很多成熟框架使用了這種寫法。
參照原專案README.md的用法範例：
numCPUs := runtime.NumCPU()
pool := tunny.NewFunc(numCPUs, func(payload interface{}) interface{} {
var result []byte
// 關心業務層的輸入、輸出即可
result = wrapSomething()
return result
})
defer pool.Close()
http.HandleFunc("/work", func(w http.ResponseWriter, r *http.Request) {
input, err := ioutil.ReadAll(r.Body)
if err != nil {
http.Error(w, "Internal error", http.StatusInternalServerError)
}
defer r.Body.Close()
// 提交任務給Process
result := pool.Process(input)
w.Write(result.([]byte))
})
http.ListenAndServe(":8080", nil)總結
Go協程有幾個內建資訊，協程id、協程棧、協程狀態(running/io wait/chan send)，通過這些資訊可以幫助我們一定程度的避免或者定位問題
Go裡面建立協程只需要一個Go關鍵字，但是要合理回收卻很關鍵，必要時可以用協程池做限制
更多程式設計相關知識，請存取：！！
以上就是聊聊Golang的協程洩露，看看怎麼預防洩露的詳細內容，更多請關注TW511.COM其它相關文章！