來，控制一下 Goroutine 的併發數量

時間 2019-11-05

標籤控制一下 goroutine 併發數量简体版

原文原文鏈接

問題

func main() {
    userCount := math.MaxInt64
    for i := 0; i < userCount; i++ {
        go func(i int) {
            // 作一些各類各樣的業務邏輯處理
            fmt.Printf("go func: %d\n", i)
            time.Sleep(time.Second)
        }(i)
    }
}

在這裏，假設 userCount 是一個外部傳入的參數（不可預測，有可能值很是大），有人會所有丟進去循環。想着所有都併發 goroutine 去同時作某一件事。以爲這樣子會效率會更高，對不對！git

那麼，你以爲這裏有沒有什麼問題？github

噩夢般的開始

固然，在特定場景下，問題可大了。由於在本文被丟進去同時併發的但是一個極端值。咱們能夠一塊兒觀察下圖的指標分析，看看狀況有多「崩潰」。下圖是上述代碼的表現：golang

輸出結果

...
go func: 5839
go func: 5840
go func: 5841
go func: 5842
go func: 5915
go func: 5524
go func: 5916
go func: 8209
go func: 8264
signal: killed

若是你本身執行過代碼，在「輸出結果」上你會遇到以下問題：服務器

系統資源佔用率不斷上漲
輸出必定數量後：控制檯就再也不刷新輸出最新的值了
信號量：signal: killed

系統負載

CPU

短期內系統負載暴增併發

虛擬內存

短期內佔用的虛擬內存暴增工具

top

PID    COMMAND      %CPU  TIME     #TH   #WQ  #PORT MEM    PURG   CMPRS  PGRP  PPID  STATE    BOOSTS
...
73414  test         100.2 01:59.50 9/1   0    18    6801M+ 0B     114G+  73403 73403 running  *0[1]

小結

若是仔細看過監控工具的示意圖，就能夠知道其實我間隔的執行了兩次，能看到系統間的使用率幅度很是大。當進程被殺掉後，總體又恢復爲正常值oop

在這裏，咱們回到主題，就是在不控制併發的 goroutine 數量 會發生什麼問題？大體以下：性能

CPU 使用率浮動上漲
Memory 佔用不斷上漲。也能夠看看 CMPRS，它表示進程的壓縮數據的字節數。已經到達 114G+ 了
主進程崩潰（被殺掉了）

簡單來講，「崩潰」的緣由就是對系統資源的佔用過大。常見的好比：打開文件數（too many files open）、內存佔用等等spa

危害

對該臺服務器產生很是大的影響，影響自身及相關聯的應用。頗有可能致使不可用或響應緩慢，另外啓動了複數「失控」的 goroutine，致使程序流轉混亂

解決方案

在前面花了大量篇幅，渲染了在存在大量併發 goroutine 數量時，不控制的話會出現「嚴重」的問題，接下來一塊兒思考下解決方案。以下：

控制/限制 goroutine 同時併發運行的數量
改變應用程序的邏輯寫法（避免大規模的使用系統資源和等待）
~~調整服務的硬件配置、最大打開數、內存等閾值~~

控制 goroutine 併發數量

接下來正式的開始解決這個問題，但願你認真閱讀的同時加以思考，由於這個問題在實際項目中真的是太常見了！

問題已經拋出來了，你須要作的是想一想有什麼辦法解決這個問題。建議你自行思考一下技術方案。再接着往下看 :-)

嘗試 chan

func main() {
    userCount := 10
    ch := make(chan bool, 2)
    for i := 0; i < userCount; i++ {
        ch <- true
        go Read(ch, i)
    }
    
    //time.Sleep(time.Second)
}

func Read(ch chan bool, i int) {
    fmt.Printf("go func: %d\n", i)
    <- ch
}

輸出結果：

go func: 1
go func: 2
go func: 3
go func: 4
go func: 5
go func: 6
go func: 7
go func: 8
go func: 0

嗯，咱們彷佛很好的控制了 2 個 2 個的「順序」執行多個 goroutine。可是，問題出現了。你仔細數一下輸出結果，才 9 個值？

這明顯就不對。緣由出在當主協程結束時，子協程也是會被終止掉的。所以剩餘的 goroutine 沒來及把值輸出，就被送上路了（不信你把 time.Sleep 打開看看，看看輸出數量）

嘗試 sync

...
var wg = sync.WaitGroup{}

func main() {
    userCount := 10
    for i := 0; i < userCount; i++ {
        wg.Add(1)
        go Read(i)
    }

    wg.Wait()
}

func Read(i int) {
    defer wg.Done()
    fmt.Printf("go func: %d\n", i)
}

嗯，單純的使用 sync.WaitGroup 也不行。沒有控制到同時併發的 goroutine 數量（代指達不到本文所要求的目標）

小結

單純簡單使用 channel 或 sync 都有明顯缺陷，不行。咱們再看看組件配合能不能實現

嘗試 chan + sync

...
var wg = sync.WaitGroup{}

func main() {
    userCount := 10
    ch := make(chan bool, 2)
    for i := 0; i < userCount; i++ {
        wg.Add(1)
        go Read(ch, i)
    }

    wg.Wait()
}

func Read(ch chan bool, i int) {
    defer wg.Done()

    ch <- true
    fmt.Printf("go func: %d, time: %d\n", i, time.Now().Unix())
    time.Sleep(time.Second)
    <-ch
}

輸出結果：

go func: 9, time: 1547911938
go func: 1, time: 1547911938
go func: 6, time: 1547911939
go func: 7, time: 1547911939
go func: 8, time: 1547911940
go func: 0, time: 1547911940
go func: 3, time: 1547911941
go func: 2, time: 1547911941
go func: 4, time: 1547911942
go func: 5, time: 1547911942

從輸出結果來看，確實實現了控制 goroutine 以 2 個 2 個的數量去執行咱們的「業務邏輯」，固然結果集也理所應當的是亂序輸出

方案一：簡單 Semaphore

在確立了簡單使用 chan + sync 的方案是可行後，咱們從新將流轉邏輯封裝爲 gsema，主程序變成以下：

import (
    "fmt"
    "time"

    "github.com/EDDYCJY/gsema"
)

var sema = gsema.NewSemaphore(3)

func main() {
    userCount := 10
    for i := 0; i < userCount; i++ {
        go Read(i)
    }

    sema.Wait()
}

func Read(i int) {
    defer sema.Done()
    sema.Add(1)

    fmt.Printf("go func: %d, time: %d\n", i, time.Now().Unix())
    time.Sleep(time.Second)
}

分析方案

在上述代碼中，程序執行流程以下：

設置容許的併發數目爲 3 個
循環 10 次，每次啓動一個 goroutine 來執行任務
每個 goroutine 在內部利用 sema 進行調控是否阻塞
按容許併發數逐漸釋出 goroutine，最後結束任務

看上去人模人樣，沒什麼嚴重問題。但卻有一個「大」坑，認真看到第二點「每次啓動一個 goroutine」這句話。這裏有點問題，提早產生那麼多的 goroutine 會不會有什麼問題，接下來一塊兒分析下利弊，以下：

利

適合量不大、複雜度低的使用場景
- 幾百幾千個、幾十萬個也是能夠接受的（看具體業務場景）
- 實際業務邏輯在運行前就已經被阻塞等待了（由於併發數受限），基本實際業務邏輯損耗的性能比 goroutine 自己大
- goroutine 自己很輕便，僅損耗極少量的內存空間和調度。這種等待響應的狀況都是躺好了，等待任務喚醒
Semaphore 操做複雜度低且流轉簡單，容易控制

弊

不適合量很大、複雜度高的使用場景
- 有幾百萬、幾千萬個 goroutine 的話，就浪費了大量調度 goroutine 和內存空間。剛好你的服務器也接受不了的話
Semaphore 操做複雜度提升，要管理更多的狀態

小結

基於什麼業務場景，就用什麼方案去作事
有足夠的時間，容許你去追求更優秀、極致的方案（用第三方庫也行）

用哪一種方案，我認爲主要基於以上兩點去思考，都是 OK 的。沒有對錯，只有當前業務場景能不能接受，這個預先啓動的 goroutine 數量你的系統是否可以接受

固然了，常見/簡單的 Go 應用採用這類技術方案，基本就能解決問題了。由於像本文第一節「問題」如此超巨大數量的狀況，狀況不多。其並不存在那些「特殊性」。所以用這個方案基本 OK

靈活控制 goroutine 併發數量

小手一緊。隔壁老王發現了新的問題。「方案一」中，在輸入輸出一體的狀況下，在常見的業務場景中確實能夠

但，此次新的業務場景比較特殊，要控制輸入的數量，以此達到改變容許併發運行 goroutine 的數量。咱們仔細想一想，要作出以下改變：

輸入/輸出要抽離，才能夠分別控制
輸入/輸出要可變，理所應當在 for-loop 中（可設置數值的地方）
容許改變 goroutine 併發數量，但它也必須有一個最大值（由於容許改變是相對）

方案二：靈活 chan + sync

package main

import (
    "fmt"
    "sync"
    "time"
)

var wg sync.WaitGroup

func main() {
    userCount := 10
    ch := make(chan int, 5)
    for i := 0; i < userCount; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            for d := range ch {
                fmt.Printf("go func: %d, time: %d\n", d, time.Now().Unix())
                time.Sleep(time.Second * time.Duration(d))
            }
        }()
    }

    for i := 0; i < 10; i++ {
        ch <- 1
        ch <- 2
        //time.Sleep(time.Second)
    }

    close(ch)
    wg.Wait()
}

輸出結果：

...
go func: 1, time: 1547950567
go func: 3, time: 1547950567
go func: 1, time: 1547950567
go func: 2, time: 1547950567
go func: 2, time: 1547950567
go func: 3, time: 1547950567
go func: 1, time: 1547950568
go func: 2, time: 1547950568
go func: 3, time: 1547950568
go func: 1, time: 1547950568
go func: 3, time: 1547950569
go func: 2, time: 1547950569

在「方案二」中，咱們能夠隨時隨地的根據新的業務需求，作以下事情：