摘要：原文地址來(lái)，控制一下的并發(fā)數(shù)量問(wèn)題做一些各種各樣的業(yè)務(wù)邏輯處理在這里，假設(shè)是一個(gè)外部傳入的參數(shù)不可預(yù)測(cè)，有可能值非常大，有人會(huì)全部丟進(jìn)去循環(huán)。因此用這個(gè)方案基本靈活控制并發(fā)數(shù)量小手一緊。

原文地址：來(lái)，控制一下 Goroutine 的并發(fā)數(shù)量

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

在這里，假設(shè) userCount 是一個(gè)外部傳入的參數(shù)（不可預(yù)測(cè)，有可能值非常大），有人會(huì)全部丟進(jìn)去循環(huán)。想著全部都并發(fā) goroutine 去同時(shí)做某一件事。覺(jué)得這樣子會(huì)效率會(huì)更高，對(duì)不對(duì)！

那么，你覺(jué)得這里有沒(méi)有什么問(wèn)題？

當(dāng)然，在特定場(chǎng)景下，問(wèn)題可大了。因?yàn)樵诒疚谋粊G進(jìn)去同時(shí)并發(fā)的可是一個(gè)極端值。我們可以一起觀察下圖的指標(biāo)分析，看看情況有多 “崩潰”。下圖是上述代碼的表現(xiàn)：

...
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

如果你自己執(zhí)行過(guò)代碼，在 “輸出結(jié)果” 上你會(huì)遇到如下問(wèn)題：

系統(tǒng)資源占用率不斷上漲

輸出一定數(shù)量后：控制臺(tái)就不再刷新輸出最新的值了

信號(hào)量：signal: killed

短時(shí)間內(nèi)系統(tǒng)負(fù)載暴增

短時(shí)間內(nèi)占用的虛擬內(nèi)存暴增

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]

如果仔細(xì)看過(guò)監(jiān)控工具的示意圖，就可以知道其實(shí)我間隔的執(zhí)行了兩次，能看到系統(tǒng)間的使用率幅度非常大。當(dāng)進(jìn)程被殺掉后，整體又恢復(fù)為正常值

在這里，我們回到主題，就是在不控制并發(fā)的 goroutine 數(shù)量 會(huì)發(fā)生什么問(wèn)題？大致如下：

CPU 使用率浮動(dòng)上漲

Memory 占用不斷上漲。也可以看看 CMPRS，它表示進(jìn)程的壓縮數(shù)據(jù)的字節(jié)數(shù)。已經(jīng)到達(dá) 114G+ 了

主進(jìn)程崩潰（被殺掉了）

簡(jiǎn)單來(lái)說(shuō)，“崩潰” 的原因就是對(duì)系統(tǒng)資源的占用過(guò)大。常見(jiàn)的比如：打開(kāi)文件數(shù)（too many files open）、內(nèi)存占用等等

對(duì)該臺(tái)服務(wù)器產(chǎn)生非常大的影響，影響自身及相關(guān)聯(lián)的應(yīng)用。很有可能導(dǎo)致不可用或響應(yīng)緩慢，另外啟動(dòng)了復(fù)數(shù) “失控” 的 goroutine，導(dǎo)致程序流轉(zhuǎn)混亂

在前面花了大量篇幅，渲染了在存在大量并發(fā) goroutine 數(shù)量時(shí)，不控制的話會(huì)出現(xiàn) “嚴(yán)重” 的問(wèn)題，接下來(lái)一起思考下解決方案。如下：

控制/限制 goroutine 同時(shí)并發(fā)運(yùn)行的數(shù)量

改變應(yīng)用程序的邏輯寫(xiě)法（避免大規(guī)模的使用系統(tǒng)資源和等待）

調(diào)整服務(wù)的硬件配置、最大打開(kāi)數(shù)、內(nèi)存等閾值

接下來(lái)正式的開(kāi)始解決這個(gè)問(wèn)題，希望你認(rèn)真閱讀的同時(shí)加以思考，因?yàn)檫@個(gè)問(wèn)題在實(shí)際項(xiàng)目中真的是太常見(jiàn)了！

問(wèn)題已經(jīng)拋出來(lái)了，你需要做的是想想有什么辦法解決這個(gè)問(wèn)題。建議你自行思考一下技術(shù)方案。再接著往下看 :-)

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
", i)
    <- ch
}

輸出結(jié)果：

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 個(gè) 2 個(gè)的 “順序” 執(zhí)行多個(gè) goroutine。但是，問(wèn)題出現(xiàn)了。你仔細(xì)數(shù)一下輸出結(jié)果，才 9 個(gè)值？

這明顯就不對(duì)。原因出在當(dāng)主協(xié)程結(jié)束時(shí)，子協(xié)程也是會(huì)被終止掉的。因此剩余的 goroutine 沒(méi)來(lái)及把值輸出，就被送上路了（不信你把 time.Sleep 打開(kāi)看看，看看輸出數(shù)量）

...
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
", i)
}

嗯，單純的使用 sync.WaitGroup 也不行。沒(méi)有控制到同時(shí)并發(fā)的 goroutine 數(shù)量（代指達(dá)不到本文所要求的目標(biāo)）

單純簡(jiǎn)單使用 channel 或 sync 都有明顯缺陷，不行。我們?cè)倏纯唇M件配合能不能實(shí)現(xiàn)

...
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
", i, time.Now().Unix())
    time.Sleep(time.Second)
    <-ch
}

輸出結(jié)果：

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

從輸出結(jié)果來(lái)看，確實(shí)實(shí)現(xiàn)了控制 goroutine 以 2 個(gè) 2 個(gè)的數(shù)量去執(zhí)行我們的 “業(yè)務(wù)邏輯”，當(dāng)然結(jié)果集也理所應(yīng)當(dāng)?shù)氖莵y序輸出

在確立了簡(jiǎn)單使用 chan + sync 的方案是可行后，我們重新將流轉(zhuǎn)邏輯封裝為 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
", i, time.Now().Unix())
    time.Sleep(time.Second)
}

在上述代碼中，程序執(zhí)行流程如下：

設(shè)置允許的并發(fā)數(shù)目為 3 個(gè)

循環(huán) 10 次，每次啟動(dòng)一個(gè) goroutine 來(lái)執(zhí)行任務(wù)

每一個(gè) goroutine 在內(nèi)部利用 sema 進(jìn)行調(diào)控是否阻塞

按允許并發(fā)數(shù)逐漸釋出 goroutine，最后結(jié)束任務(wù)

看上去人模人樣，沒(méi)什么嚴(yán)重問(wèn)題。但卻有一個(gè) “大” 坑，認(rèn)真看到第二點(diǎn) “每次啟動(dòng)一個(gè) goroutine” 這句話。這里有點(diǎn)問(wèn)題，提前產(chǎn)生那么多的 goroutine 會(huì)不會(huì)有什么問(wèn)題，接下來(lái)一起分析下利弊，如下：

適合量不大、復(fù)雜度低的使用場(chǎng)景

幾百幾千個(gè)、幾十萬(wàn)個(gè)也是可以接受的（看具體業(yè)務(wù)場(chǎng)景）

實(shí)際業(yè)務(wù)邏輯在運(yùn)行前就已經(jīng)被阻塞等待了（因?yàn)椴l(fā)數(shù)受限），基本實(shí)際業(yè)務(wù)邏輯損耗的性能比 goroutine 本身大

goroutine 本身很輕便，僅損耗極少許的內(nèi)存空間和調(diào)度。這種等待響應(yīng)的情況都是躺好了，等待任務(wù)喚醒

Semaphore 操作復(fù)雜度低且流轉(zhuǎn)簡(jiǎn)單，容易控制

不適合量很大、復(fù)雜度高的使用場(chǎng)景

有幾百萬(wàn)、幾千萬(wàn)個(gè) goroutine 的話，就浪費(fèi)了大量調(diào)度 goroutine 和內(nèi)存空間。恰好你的服務(wù)器也接受不了的話

Semaphore 操作復(fù)雜度提高，要管理更多的狀態(tài)

基于什么業(yè)務(wù)場(chǎng)景，就用什么方案去做事

有足夠的時(shí)間，允許你去追求更優(yōu)秀、極致的方案（用第三方庫(kù)也行）

用哪種方案，我認(rèn)為主要基于以上兩點(diǎn)去思考，都是 OK 的。沒(méi)有對(duì)錯(cuò)，只有當(dāng)前業(yè)務(wù)場(chǎng)景能不能接受，這個(gè)預(yù)先啟動(dòng)的 goroutine 數(shù)量你的系統(tǒng)是否能夠接受

當(dāng)然了，常見(jiàn)/簡(jiǎn)單的 Go 應(yīng)用采用這類技術(shù)方案，基本就能解決問(wèn)題了。因?yàn)橄癖疚牡谝还?jié) “問(wèn)題” 如此超巨大數(shù)量的情況，情況很少。其并不存在那些 “特殊性”。因此用這個(gè)方案基本 OK

小手一緊。隔壁老王發(fā)現(xiàn)了新的問(wèn)題?！胺桨敢弧?中，在輸入輸出一體的情況下，在常見(jiàn)的業(yè)務(wù)場(chǎng)景中確實(shí)可以

但，這次新的業(yè)務(wù)場(chǎng)景比較特殊，要控制輸入的數(shù)量，以此達(dá)到改變?cè)试S并發(fā)運(yùn)行 goroutine 的數(shù)量。我們仔細(xì)想想，要做出如下改變：

輸入/輸出要抽離，才可以分別控制

輸入/輸出要可變，理所應(yīng)當(dāng)在 for-loop 中（可設(shè)置數(shù)值的地方）

允許改變 goroutine 并發(fā)數(shù)量，但它也必須有一個(gè)最大值（因?yàn)樵试S改變是相對(duì)）

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
", 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()
}

輸出結(jié)果：

...
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

在 “方案二” 中，我們可以隨時(shí)隨地的根據(jù)新的業(yè)務(wù)需求，做如下事情：

變更 channel 的輸入數(shù)量

能夠根據(jù)特殊情況，變更 channel 的循環(huán)值

變更最大允許并發(fā)的 goroutine 數(shù)量

總的來(lái)說(shuō)，就是可控空間都盡量放開(kāi)了，是不是更加靈活了呢 :-)

go-playground/pool

nozzle/throttler

Jeffail/tunny

panjf2000/ants

比較成熟的第三方庫(kù)也不少，基本都是以生成和管理 goroutine 為目標(biāo)的池工具。我簡(jiǎn)單列了幾個(gè)，具體建議大家閱讀下源碼或者多找找，原理相似

在本文的開(kāi)頭，我花了大力氣（極端數(shù)量），告訴你同時(shí)并發(fā)過(guò)多的 goroutine 數(shù)量會(huì)導(dǎo)致系統(tǒng)占用資源不斷上漲。最終該服務(wù)崩盤(pán)的極端情況。為的是希望你今后避免這種問(wèn)題，給你留下深刻的印象

接下來(lái)我們以 “控制 goroutine 并發(fā)數(shù)量” 為主題，展開(kāi)了一番分析。分別給出了三種方案。在我看來(lái)，各具優(yōu)缺點(diǎn)，我建議你挑選合適自身場(chǎng)景的技術(shù)方案就可以了

因?yàn)?，有不同類型的技術(shù)方案也能解決這個(gè)問(wèn)題，千人千面。本文推薦的是較常見(jiàn)的解決方案，也歡迎大家在評(píng)論區(qū)繼續(xù)補(bǔ)充 :-)