快捷導航

Golang并發(fā)控制的三種實現(xiàn)方法

更新時間：2023年08月23日 11:04:25 作者：林林林ZEYU

在Golang中,有多種方式可以進行并發(fā)控制,本文詳細的介紹了三種實現(xiàn)方法,Channel優(yōu)點是實現(xiàn)簡單,清晰易懂,WaitGroup優(yōu)點是子協(xié)程個數(shù)動態(tài)可調(diào)整,Context 優(yōu)點是對子協(xié)程派生出來的孫子協(xié)程的控制,缺點是相對而言的,要結(jié)合實例應(yīng)用場景進行選擇

三種方案各有優(yōu)劣，比如Channel優(yōu)點是實現(xiàn)簡單，清晰易懂，WaitGroup優(yōu)點是子協(xié)程個數(shù)動態(tài)可調(diào)整，Context 優(yōu)點是對子協(xié)程派生出來的孫子協(xié)程的控制。缺點是相對而言的，要結(jié)合實例應(yīng)用場景進行選擇。

channel

channel一般用于協(xié)程之間的通信，channel也可以用于并發(fā)控制。比如主協(xié)程啟動N個子協(xié)程，主協(xié)程等待所有子協(xié)程退出后再繼續(xù)后續(xù)流程，這種場景下channel也可輕易實現(xiàn)。

場景示例

下面程序展示一個使用channel控制子協(xié)程的例子：

上面程序通過創(chuàng)建N個channel來管理N個協(xié)程，每個協(xié)程都有一個channel用于跟父協(xié)程通信，父協(xié)程創(chuàng)建完所有協(xié) 程中等待所有協(xié)程結(jié)束。這個例子中，父協(xié)程僅僅是等待子協(xié)程結(jié)束，其實父協(xié)程也可以向管道中寫入數(shù)據(jù)通知子協(xié)程結(jié)束，這時子協(xié)程需要定期的探測管道中是否有消息出現(xiàn)。

總結(jié)

使用channel來控制子協(xié)程的優(yōu)點是實現(xiàn)簡單，缺點是當需要大量創(chuàng)建協(xié)程時就需要有相同數(shù)量的channel，而且對于子協(xié)程繼續(xù)派生出來的協(xié)程不方便控制。后面繼續(xù)介紹的WaitGroup、Context看起來比channel優(yōu)雅一些，在各種開源組件中使用頻率比channel高得多。

WaitGroup

WaitGroup是Golang應(yīng)用開發(fā)過程中經(jīng)常使用的并發(fā)控制技術(shù)。 WaitGroup，可理解為Wait-Goroutine-Group，即等待一組goroutine結(jié)束。比如某個goroutine需要等待其他幾個goroutine全部完成，那么使用WaitGroup可以輕松實現(xiàn)。

下面程序展示了一個goroutine等待另外兩個goroutine結(jié)束的例子：

簡單的說，上面程序中wg內(nèi)部維護了一個計數(shù)器：

1.啟動goroutine前將計數(shù)器通過Add(2)將計數(shù)器設(shè)置為待啟動的goroutine個數(shù)。
2. 啟動goroutine后，使用Wait()方法阻塞自己，等待計數(shù)器變?yōu)?。
3. 每個goroutine執(zhí)行結(jié)束通過Done()方法將計數(shù)器減1。
4. 計數(shù)器變?yōu)?后，阻塞的goroutine被喚醒。

其實WaitGroup也可以實現(xiàn)一組goroutine等待另一組goroutine，這有點像玩雜技，很容出錯，如果不了解其實現(xiàn)原理更是如此。實際上，WaitGroup的實現(xiàn)源碼非常簡單。

信號量

信號量是Unix系統(tǒng)提供的一種保護共享資源的機制，用于防止多個線程同時訪問某個資源。可簡單理解為信號量為一個數(shù)值：
當信號量>0時，表示資源可用，獲取信號量時系統(tǒng)自動將信號量減1；
當信號量= =0時，表示資源暫不可用，獲取信號量時，當前線程會進入睡眠，當信號量為正時被喚醒；
由于WaitGroup實現(xiàn)中也使用了信號量，在此做個簡單介紹。

WaitGroup數(shù)據(jù)結(jié)構(gòu)

源碼包中 src/sync/waitgroup.go:WaitGroup 定義了其數(shù)據(jù)結(jié)構(gòu)：

type WaitGroup struct { 
state1 [3]uint32
}

state1是個長度為3的數(shù)組，其中包含了state和一個信號量，而state實際上是兩個計數(shù)器：

counter：當前還未執(zhí)行結(jié)束的goroutine計數(shù)器
waiter count: 等待goroutine-group結(jié)束的goroutine數(shù)量，即有多少個等候者 semaphore: 信號量

考慮到字節(jié)是否對齊，三者出現(xiàn)的位置不同，為簡單起見，依照字節(jié)已對齊情況下，三者在內(nèi)存中的位置如下所示：

WaitGroup對外提供三個接口：

Add(delta int): 將delta值加到counter中
Wait()： waiter遞增1，并阻塞等待信號量
semaphore Done()： counter遞減1，按照waiter數(shù)值釋放相應(yīng)次數(shù)信號量

下面分別介紹這三個函數(shù)的實現(xiàn)細節(jié)。

Add(delta int)

Add()做了兩件事，一是把delta值累加到counter中，因為delta可以為負值，也就是說counter有可能變成0或負值，所以第二件事就是當counter值變?yōu)?時，跟據(jù)waiter數(shù)值釋放等量的信號量，把等待的goroutine全部喚醒，如果counter變?yōu)樨撝担瑒tpanic

Add()偽代碼如下：

Wait()

Wait()方法也做了兩件事，一是累加waiter, 二是阻塞等待信號量

這里用到了CAS算法保證有多個goroutine同時執(zhí)行Wait()時也能正確累加waiter。

Done()

Done()只做一件事，即把counter減1，我們知道Add()可以接受負值，所以Done實際上只是調(diào)用了Add(-1)。源碼如下：

func (wg *WaitGroup) Done() { 
wg.Add(-1) 
}

Done()的執(zhí)行邏輯就轉(zhuǎn)到了Add()，實際上也正是最后一個完成的goroutine把等待者喚醒的。

注意事項：Add()操作必須早于Wait(), 否則會panicAdd()設(shè)置的值必須與實際等待的goroutine個數(shù)一致，否則會panic

context

Golang context是Golang應(yīng)用開發(fā)常用的并發(fā)控制技術(shù)，它與WaitGroup最大的不同點是context對于派生 goroutine有更強的控制力，它可以控制多級的goroutine。
context翻譯成中文是”上下文”，即它可以控制一組呈樹狀結(jié)構(gòu)的goroutine，每個goroutine擁有相同的上下文。

典型的使用場景如下圖所示：

上圖中由于goroutine派生出子goroutine，而子goroutine又繼續(xù)派生新的goroutine，這種情況下使用 WaitGroup就不太容易，因為子goroutine個數(shù)不容易確定。而使用context就可以很容易實現(xiàn)。

Context實現(xiàn)原理

context實際上只定義了接口，凡是實現(xiàn)該接口的類都可稱為是一種context，官方包中實現(xiàn)了幾個常用的 context，分別可用于不同的場景。

接口定義

源碼包中 src/context/context.go:Context 定義了該接口：

基礎(chǔ)的context接口只定義了4個方法，下面分別簡要說明一下：

Deadline()

該方法返回一個deadline和標識是否已設(shè)置deadline的bool值，如果沒有設(shè)置deadline，則ok == false，此時deadline為一個初始值的time.Time值

Done()

該方法返回一個channel，需要在select-case語句中使用，如”case <-context.Done():”。
當context關(guān)閉后，Done()返回一個被關(guān)閉的管道，關(guān)閉的管理仍然是可讀的，據(jù)此goroutine可以收到關(guān)閉請求；當context還未關(guān)閉時，Done()返回nil。

Err()

該方法描述context關(guān)閉的原因。關(guān)閉原因由context實現(xiàn)控制，不需要用戶設(shè)置。比如Deadline context，關(guān) 閉原因可能是因為deadline，也可能提前被主動關(guān)閉，那么關(guān)閉原因就會不同:
因deadline關(guān)閉：“context deadline exceeded”；
因主動關(guān)閉： “context canceled”。
當context關(guān)閉后，Err()返回context的關(guān)閉原因；當context還未關(guān)閉時，Err()返回nil；

Value()

有一種context，它不是用于控制呈樹狀分布的goroutine，而是用于在樹狀分布的goroutine間傳遞信息。
Value()方法就是用于此種類型的context，該方法根據(jù)key值查詢map中的value。具體使用后面示例說明。

空context

context包中定義了一個空的context，名為emptyCtx，用于context的根節(jié)點，空的context只是簡單的實現(xiàn) 了Context，本身不包含任何值，僅用于其他context的父節(jié)點。
emptyCtx類型定義如下代碼所示：

context包中定義了一個公用的emptCtx全局變量，名為background，可以使用context.Background()獲取它，實現(xiàn)代碼如下所示：

context包提供了4個方法創(chuàng)建不同類型的context，使用這四個方法時如果沒有父context，都需要傳入 backgroud，即backgroud作為其父節(jié)點：

WithCancel()
WithDeadline()
WithTimeout()
WithValue()

context包中實現(xiàn)Context接口的struct，除了emptyCtx外，還有cancelCtx、timerCtx和valueCtx三種，正是基于這三種context實例，實現(xiàn)了上述4種類型的context。

context包中各context類型之間的關(guān)系，如下圖所示：

struct cancelCtx、valueCtx、valueCtx都繼承于Context，下面分別介紹這三個struct。

cancelCtx

源碼包中 src/context/context.go:cancelCtx 定義了該類型context：

children中記錄了由此context派生的所有child，此context被cancle時會把其中的所有child都cancle掉。

cancelCtx與deadline和value無關(guān)，所以只需要實現(xiàn)Done()和Err()接口外露接口即可。

Done()接口實現(xiàn)

按照Context定義，Done()接口只需要返回一個channel即可，對于cancelCtx來說只需要返回成員變量done即可。

這里直接看下源碼，非常簡單：

由于cancelCtx沒有指定初始化函數(shù)，所以cancelCtx.done可能還未分配，所以需要考慮初始化。 cancelCtx.done會在context被cancel時關(guān)閉，所以cancelCtx.done的值一般經(jīng)歷如三個階段：nil —> chan struct{} —> closed chan。

Err()接口實現(xiàn)

按照Context定義，Err()只需要返回一個error告知context被關(guān)閉的原因。對于cancelCtx來說只需要返回成員變量err即可。還是直接看下源碼：

cancelCtx.err默認是nil，在context被cancel時指定一個error變量： var Canceled = errors.New(“context canceled”) 。

cancel()接口實現(xiàn)

cancel()內(nèi)部方法是理解cancelCtx的最關(guān)鍵的方法，其作用是關(guān)閉自己和其后代，其后代存儲在 cancelCtx.children的map中，其中key值即后代對象，value值并沒有意義，這里使用map只是為了方便查詢而已。
cancel方法實現(xiàn)偽代碼如下所示：

實際上，WithCancel()返回的第二個用于cancel context的方法正是此cancel()。

WithCancel()方法實現(xiàn)

WithCancel()方法作了三件事：

初始化一個cancelCtx實例
將cancelCtx實例添加到其父節(jié)點的children中(如果父節(jié)點也可以被cancel的話)
返回cancelCtx實例和cancel()方法

其實現(xiàn)源碼如下所示：

這里將自身添加到父節(jié)點的過程有必要簡單說明一下：

如果父節(jié)點也支持cancel，也就是說其父節(jié)點肯定有children成員，那么把新context添加到children里即可；
如果父節(jié)點不支持cancel，就繼續(xù)向上查詢，直到找到一個支持cancel的節(jié)點，把新context添加到 children里；
如果所有的父節(jié)點均不支持cancel，則啟動一個協(xié)程等待父節(jié)點結(jié)束，然后再把當前context結(jié)束。

典型使用案例

一個典型的使用cancel context的例子如下所示：

上面代碼中協(xié)程HandelRequest()用于處理某個請求，其又會創(chuàng)建兩個協(xié)程：WriteRedis()、 WriteDatabase()，main協(xié)程創(chuàng)建創(chuàng)建context，并把context在各子協(xié)程間傳遞，main協(xié)程在適當?shù)臅r機可以 cancel掉所有子協(xié)程。

程序輸出如下所示：

timerCtx

源碼包中 src/context/context.go:timerCtx 定義了該類型context：

type timerCtx struct { 
cancelCtx 
timer *time.Timer // Under cancelCtx.mu. 
deadline time.Time 
}

timerCtx在cancelCtx基礎(chǔ)上增加了deadline用于標示自動cancel的最終時間，而timer就是一個觸發(fā)自動 cancel的定時器。由此，衍生出WithDeadline()和WithTimeout()。實現(xiàn)上這兩種類型實現(xiàn)原理一樣，只不過使用語境不一樣：deadline: 指定最后期限，比如context將2018.10.20 00:00:00之時自動結(jié)束 timeout: 指定最長存活時間，比如context將在30s后結(jié)束。對于接口來說，timerCtx在cancelCtx基礎(chǔ)上還需要實現(xiàn)Deadline()和cancel()方法，其中cancel()方法是重寫的。

Deadline()接口實現(xiàn)

Deadline()方法僅僅是返回timerCtx.deadline而矣。而timerCtx.deadline是WithDeadline()或 WithTimeout()方法設(shè)置的。

cancel()接口實現(xiàn)

cancel()方法基本繼承cancelCtx，只需要額外把timer關(guān)閉。
timerCtx被關(guān)閉后，timerCtx.cancelCtx.err將會存儲關(guān)閉原因：
如果deadline到來之前手動關(guān)閉，則關(guān)閉原因與cancelCtx顯示一致；
如果deadline到來時自動關(guān)閉，則原因為：”context deadline exceeded”

WithDeadline()方法實現(xiàn)

WithDeadline()方法實現(xiàn)步驟如下：
初始化一個timerCtx實例
將timerCtx實例添加到其父節(jié)點的children中(如果父節(jié)點也可以被cancel的話)
啟動定時器，定時器到期后會自動cancel本context
返回timerCtx實例和cancel()方法

也就是說，timerCtx類型的context不僅支持手動cancel，也會在定時器到來后自動cancel。

WithTimeout()方法實現(xiàn)

WithTimeout()實際調(diào)用了WithDeadline，二者實現(xiàn)原理一致。看代碼會非常清晰：

func WithTimeout(parent Context, timeout time.Duration) (Context, CancelFunc) { 
return WithDeadline(parent, time.Now().Add(timeout)) 
}

典型使用案例下面例子中使用WithTimeout()獲得一個context并在其了協(xié)程中傳遞：

主協(xié)程中創(chuàng)建一個10s超時的context，并將其傳遞給子協(xié)程，10s自動關(guān)閉context。程序輸出如下：

valueCtx

源碼包中 src/context/context.go:valueCtx 定義了該類型context：

type valueCtx struct { 
Context 
key, val interface{} 
}

valueCtx只是在Context基礎(chǔ)上增加了一個key-value對，用于在各級協(xié)程間傳遞一些數(shù)據(jù)。由于valueCtx既不需要cancel，也不需要deadline，那么只需要實現(xiàn)Value()接口即可。

Value（）接口實現(xiàn)

由valueCtx數(shù)據(jù)結(jié)構(gòu)定義可見，valueCtx.key和valueCtx.val分別代表其key和value值。實現(xiàn)也很簡單：

這里有個細節(jié)需要關(guān)注一下，即當前context查找不到key時，會向父節(jié)點查找，如果查詢不到則最終返回interface{}。也就是說，可以通過子context查詢到父的value值。

WithValue（）方法實現(xiàn)

WithValue()實現(xiàn)也是非常的簡單, 偽代碼如下：

典型使用案例

下面示例程序展示valueCtx的用法：

上例main()中通過WithValue()方法獲得一個context，需要指定一個父context、key和value。然后通將該 context傳遞給子協(xié)程HandelRequest，子協(xié)程可以讀取到context的key-value。注意：本例中子協(xié)程無法自動結(jié)束，因為context是不支持cancle的，也就是說<-ctx.Done()永遠無法返回。

如果需要返回，需要在創(chuàng)建context時指定一個可以cancel的context作為父節(jié)點，使用父節(jié)點的cancel()在適當?shù)?時機結(jié)束整個context。