【原】goroutine和channel 如何控制并發(fā)順序？

小生凡一 2022-03-31

展開全文

文章目錄

寫在前面

最近有同學問我這個問題。

在這里插入圖片描述

題目意思是利用goroutine和channel 連續(xù)輸出10次，dog，cat，fish，并且都要按照這個dog，cat，fish的順序輸出。

分析

題目既然要求是使用goroutine，那么我們肯定是要控制好這個并發(fā)的順序。因為并發(fā)是具有隨機性的，這個題目并不難，很典型的chan控制進程之間的順序。

那我們先了解一下 goroutine ，select ，sync.WaitGroup，channel

1. goroutine

我們這里先了解一下go的調(diào)度機制，即是GPM模型。goruntine相對線程更加輕量，GPM調(diào)度器效率更高。

G：Goroutine 我們所說的協(xié)程，為用戶級的輕量級線程，每個Goroutine對象中的sched保存著其上下文信息
P：Processor 調(diào)度，即為G和M的調(diào)度對象，用來調(diào)度G和M之間的關(guān)聯(lián)關(guān)系，其數(shù)量可通過GOMAXPROCS()來設(shè)置，默認為核心數(shù)
M：Machine 真正的工人，對內(nèi)核級線程的封裝，數(shù)量對應(yīng)真實的CPU數(shù)

每個Processor對象都擁有一個LRQ（Local Run Queue），未分配的Goroutine對象保存在GRQ（Global Run Queue ）中，等待分配給某一個P的LRQ中，每個LRQ里面包含若干個用戶創(chuàng)建的Goroutine對象。

同時Processor作為橋梁對Machine和Goroutine進行了解耦，也就是說Goroutine如果想要使用Machine需要綁定一個Processor才行。

2. select

select 和 switch 很像，它不需要輸入?yún)?shù)，并且僅僅被使用在通道操作上。
select 語句被用來執(zhí)行多個通道操作的一個和其附帶的 case 塊代碼。

我們知道 select 語句和 switch 很像，不同點是用通道讀寫操作代替了布爾操作。
通道將被阻塞，除非它有默認的 default 塊 (之后將介紹)。一旦某個 case 條件執(zhí)行，它將不阻塞。
我們發(fā)現(xiàn) select 語句將阻塞，因此 select 將等待，直到有 case 語句不阻塞。

可以使用 select 模擬了一個數(shù)百萬請求的服務(wù)器負載均衡的例子，它從多個有效服務(wù)中返回其中一個響應(yīng)。

使用協(xié)程，通道和 select 語句，我們可以向多個服務(wù)器請求數(shù)據(jù)并獲取其中最快響應(yīng)的那個。

3. sync.WaitGroup

WaitGroup 是一個帶著計數(shù)器的結(jié)構(gòu)體，這個計數(shù)器可以追蹤到有多少協(xié)程創(chuàng)建，有多少協(xié)程完成了其工作。當計數(shù)器為 0 的時候說明所有協(xié)程都完成了其工作。

Add 方法的參數(shù)是一個變量名叫 delta 的int 類型參數(shù)，主要用來內(nèi)部計數(shù)。內(nèi)部計數(shù)器默認值為 0. 它用于記錄多少個協(xié)程在運行。
當 WaitGroup 創(chuàng)建后，計數(shù)器值為 0，我們可以通過給 Add方法傳 int類型值來增加它的數(shù)量。記住，當協(xié)程建立后，計數(shù)器的值不會自動遞增，因此需要我們手動遞增它。
Wait 方法用來阻塞當前協(xié)程。一旦計數(shù)器為 0, 協(xié)程將恢復(fù)運行。因此，我們需要一個方法去降低計數(shù)器的值。
Done 方法可以降低計數(shù)器的值。他不接受任何參數(shù)，因此，它每執(zhí)行一次計數(shù)器就減 1。

4. channel

channel 具體看這篇文章吧 channel介紹

之前的一篇博客已經(jīng)講的很清楚了。

5. 代碼

簡單了解完上述之后，我們開始寫代碼。

解釋

既然是并發(fā)，那么我們就要寫3個函數(shù)，去分別打印我們的dog，cat，fish了。

這里用dog進行舉例

func dog(){
fmt.Println("dog")
}

那我們的主函數(shù)就要啟動goroutine去并發(fā)了。大概就是一下這種情況。

func main(){
//...省略一些邏輯
go dog()
go cat()
go fish()
//...省略一些邏輯
}

那么我們先控制這三個的并發(fā)順序，可以直接select去阻塞進行調(diào)試。

既然要控制并發(fā)順序，我們就要可以用channel進行通信通知。我們先創(chuàng)建三個channel，用chan去傳遞信息。注意這里是傳遞無緩沖的channel，因為無緩沖是可以進行讀寫同步的。用來控制并發(fā)順序最合適不過了。

dogChan, catChan, fishChan := make(chan bool), make(chan bool), make(chan bool)

dogChan 一開始賦值，并且dog打印完之后，給catChan通信，cat打印完之后，給fishChan通信，fish打印完后給dogChan通信。打完10次之后就停止。

比如這個傳入dogChan 和 catChan 進行通信。把dogChan的取出，再將catChan的賦值，就可以不斷進行循環(huán)調(diào)度了。

func dog(dogChan chan bool,catChan chan bool ) {
for {
select {
case <-dogChan:
fmt.Println("dog")
catChan <- true
break
default:
break
}
}
}

我們主程序可以用 sync.WaitGroup 來進行阻塞。當完成10次之后才Done掉，那么就完成了。

func fish(fishChan chan bool,dogChan chan bool ) {
i := 0
for {
select {
case <-fishChan:
fmt.Println("fish")
i++
if i > 9 {
wg.Done()
return
}
dogChan <- true
break
default:
break
}
}
}

完整

package main

import (
"fmt"
"sync"
)

var wg sync.WaitGroup

func dog(dogChan chan bool,catChan chan bool ) {
for {
select {
case <-dogChan:
fmt.Println("dog")
catChan <- true
break
default:
break
}
}
}

func cat(catChan chan bool,fishChan chan bool ) {
for {
select {
case <-catChan:
fmt.Println("cat")
fishChan <- true
break
default:
break
}
}
}

func fish(fishChan chan bool,dogChan chan bool ) {
i := 0
for {
select {
case <-fishChan:
fmt.Println("fish")
i++ // 計數(shù)，打印完之后就溜溜結(jié)束了。
if i > 9 {
wg.Done()
return
}
dogChan <- true
break
default:
break
}
}
}

func main() {
dogChan, catChan, fishChan := make(chan bool), make(chan bool), make(chan bool)
wg.Add(1)
go dog(dogChan, catChan)
go cat(catChan, fishChan)
go fish(fishChan, dogChan)
dogChan <- true // 記得這里進行啟動條件，不然就沒法啟動了。
wg.Wait()
}