go語(yǔ)言中的協(xié)程詳解

更新時(shí)間：2022年07月16日 09:05:44 作者：奮斗的大橙子

本文詳細(xì)講解了go語(yǔ)言中的協(xié)程，文中通過(guò)示例代碼介紹的非常詳細(xì)。對(duì)大家的學(xué)習(xí)或工作具有一定的參考借鑒價(jià)值，需要的朋友可以參考下

協(xié)程的特點(diǎn)

1.該任務(wù)的業(yè)務(wù)代碼主動(dòng)要求切換，即主動(dòng)讓出執(zhí)行權(quán)限

2.發(fā)生了IO，導(dǎo)致執(zhí)行阻塞（使用channel讓協(xié)程阻塞）

與線(xiàn)程本質(zhì)的不同

C#、java中我們執(zhí)行多個(gè)線(xiàn)程，是通過(guò)時(shí)間片切換來(lái)進(jìn)行的，要知道進(jìn)行切換，程序需要保存上下文等信息，是比較消耗性能的

GO語(yǔ)言中的協(xié)程，沒(méi)有上面這種切換，一定是通過(guò)協(xié)程主動(dòng)放出權(quán)限，不是被動(dòng)的。

例如：

C# 中創(chuàng)建兩個(gè)線(xiàn)程

可以看到1和2是交替執(zhí)行的

Go語(yǔ)言中用協(xié)程實(shí)現(xiàn)一下

runtime.GOMAXPROCS(1)

這個(gè)結(jié)果就是執(zhí)行了1 在執(zhí)行2

上述兩種方式來(lái)進(jìn)行協(xié)程的切換

1.該任務(wù)的業(yè)務(wù)代碼主動(dòng)要求切換，即主動(dòng)讓出執(zhí)行權(quán)限

2.通過(guò)channel進(jìn)行阻塞

執(zhí)行的結(jié)果一樣

后面又繼續(xù)執(zhí)行1了

一、Goroutine的并行

package main

import (
　　"fmt"
　　"runtime"
　　"sync"
)

func main() {
　　runtime.GOMAXPROCS(1)
　　var wg sync.WaitGroup
　　wg.Add(3)
　　fmt.Println("Start Goroutines")

　　go func() {
　　　　defer wg.Done()
　　　　for i:=1;i<=10;i++{
　　　　　　fmt.Print("1")
　　　　}
　　}()

　　go func() {
　　　　defer wg.Done()
　　　　for i:=1;i<=10;i++ {
            fmt.Print("2")
　　　　}
　　}()

　　go func() {
　　　　defer wg.Done()
　　　　for i:=1;i<=10;i++ {
　　         fmt.Print("3")
　　　　}
　　}()

　　fmt.Println("等待執(zhí)行結(jié)束")
　　wg.Wait()
}

運(yùn)行結(jié)果

Start Goroutines

等待執(zhí)行結(jié)束

333333333311111111112222222222

（1）runtime.GOMAXPROCS(1) 的作用是什么？

runtime包的GOMAXPROCS 函數(shù)。這個(gè)函數(shù)允許程序更改調(diào)度器可以使用的邏輯處理器的數(shù)量。如果不想在代碼里做這個(gè)調(diào)用，也可以通過(guò)修改和這個(gè)函數(shù)名字一樣的環(huán)境變量的值來(lái)更改邏輯處理器的數(shù)量。

（2）為什么先輸出3后輸出1和2？

調(diào)度器源碼中有體現(xiàn)：有一個(gè)隊(duì)列，還有一個(gè)優(yōu)先執(zhí)行——即最后一個(gè)，在只有一個(gè)邏輯處理器的情況下，先執(zhí)行優(yōu)先的那個(gè)，再順序執(zhí)行隊(duì)列中的。不過(guò)經(jīng)過(guò)各種實(shí)驗(yàn)發(fā)現(xiàn)這個(gè)執(zhí)行順序其實(shí)是變的，貌似糾結(jié)這個(gè)順序也沒(méi)有什么意思。

（3）WaitGroup 是什么意思？

WaitGroup 是一個(gè)計(jì)數(shù)信號(hào)量，可以用來(lái)記錄并維護(hù)運(yùn)行的goroutine。如果WaitGroup的值大于0，Wait 方法就會(huì)阻塞。為了減小WaitGroup 的值并最終釋放main 函數(shù)，使用defer 聲明在函數(shù)退出時(shí)

調(diào)用Done 方法。（defer 有點(diǎn)像C#當(dāng)中的fianlly）

補(bǔ)充：調(diào)度算法中，如果多個(gè)goroutine中某個(gè)執(zhí)行過(guò)長(zhǎng)，此時(shí)會(huì)將其停止讓給其他goroutine繼續(xù)執(zhí)行，待到其他都執(zhí)行完成，在將其進(jìn)行執(zhí)行。如下圖：G4和G5就進(jìn)行了切換

如下代碼可以驗(yàn)證上面的問(wèn)題：

package main

import (
    "fmt"
    "runtime"
    "sync"
)

var wg sync.WaitGroup

func main() {
    runtime.GOMAXPROCS(1)

    wg.Add(2)

    fmt.Println("Create Goroutines")
    go printPrime("A")
    go printPrime("B")

    fmt.Println("Waiting To Finish")
    wg.Wait()

    fmt.Println("Terminating Program")
}


func printPrime(prefix string) {
    defer wg.Done()

next:
    for outer := 2; outer < 5000; outer++ {
        for inner := 2; inner < outer; inner++ {
            if outer%inner == 0 {
                continue next
            }
        }
        fmt.Printf("%s:%d\n", prefix, outer)
    }
    fmt.Println("Completed", prefix)
}

printPrime 這個(gè)函數(shù)作用是查找顯示 5000 以?xún)?nèi)的素?cái)?shù)值，這是一個(gè)比較耗時(shí)的程序。

運(yùn)行結(jié)果：數(shù)字比較多，不打印了，但是可以看到結(jié)果是A和B兩個(gè)協(xié)程之間的切換。

上面的代碼都是設(shè)置GOMAXPROCS為1的情況，給每個(gè)可用的核心分配一個(gè)邏輯處理器

runtime.GOMAXPROCS(runtime.NumCPU())

這樣運(yùn)行第一個(gè)程序的結(jié)果如下

Start Goroutines
等待執(zhí)行結(jié)束133333333333333333333333333333333333333333333333333333333333333333333333333333333333333333333333333332111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111222222222222222222222222222222222222222222222222222222222222222222222222222222222222222222222222222

可以看到1、2、3沒(méi)有什么規(guī)律。

二、Goroutine的競(jìng)爭(zhēng)

什么競(jìng)爭(zhēng)狀態(tài)：如果兩個(gè)或者多個(gè)goroutine 在沒(méi)有互相同步的情況下，訪(fǎng)問(wèn)某個(gè)共享的資源，并試圖同時(shí)讀和寫(xiě)這個(gè)資源，就處于相互競(jìng)爭(zhēng)的狀態(tài)，這種情況被稱(chēng)作競(jìng)爭(zhēng)狀態(tài)（race candition）

競(jìng)爭(zhēng)狀態(tài)是比較容易出現(xiàn)問(wèn)題的地方，所以對(duì)一個(gè)共享資源的讀和寫(xiě)操作必須是原子化的，換句話(huà)說(shuō)，同一時(shí)刻只能有一個(gè)goroutine 對(duì)共享資源進(jìn)行讀和寫(xiě)操作。

package main

import (
    "fmt"
    "runtime"
    "sync"
)

var (
    counter int
    wg sync.WaitGroup
)

func main() {
    wg.Add(2)

    go incCounter(1)
    go incCounter(2)

    wg.Wait()
    fmt.Println("Final Counter:", counter)
}

func incCounter(id int) {
    defer wg.Done()
    for count := 0; count < 2; count++ {
        value := counter
        runtime.Gosched()
        value++
        counter = value
    }
}

運(yùn)行結(jié)果：

Final Counter: 2

變量 counter 會(huì)進(jìn)行4 次讀和寫(xiě)操作，每個(gè)goroutine 執(zhí)行兩次。但是，程序終止時(shí)，counter變量的值為2，這是因?yàn)閮蓚€(gè)協(xié)程之間產(chǎn)生了競(jìng)爭(zhēng)，對(duì)同一個(gè)counter進(jìn)行讀寫(xiě)，下面這個(gè)圖很好的詮釋了為什么最后counter是2

補(bǔ)充：runtime.Gosched()用于讓出CPU時(shí)間片。

三、使用Goroutine的鎖

一種修正代碼、消除競(jìng)爭(zhēng)狀態(tài)的辦法是，使用Go 語(yǔ)言提供的鎖機(jī)制，來(lái)鎖住共享資源，

從而保證goroutine 的同步狀態(tài)。Go 語(yǔ)言提供了傳統(tǒng)的同步goroutine 的機(jī)制，就是對(duì)共享資源加鎖。

1、原子函數(shù)

使用atomic包來(lái)提供對(duì)數(shù)值類(lèi)型的安全訪(fǎng)問(wèn)。

package main

import (
    "fmt"
    "runtime"
    "sync"
    "sync/atomic"
)

var (
    counter int64
    wg sync.WaitGroup
)

func main() {
    wg.Add(2)

    go incCounter(1)
    go incCounter(2)

    wg.Wait()

    fmt.Println("Final Counter:", counter)
}

func incCounter(id int) {
    defer wg.Done()
    for count := 0; count < 2; count++ {
        atomic.AddInt64(&counter, 1)
        runtime.Gosched()
    }
}

運(yùn)行結(jié)果

Final Counter: 4

atmoic 包的AddInt64 函數(shù)。這個(gè)函數(shù)會(huì)同步整型值的加法，方法是強(qiáng)制同一時(shí)刻只能有一個(gè)goroutine 運(yùn)行并完成這個(gè)加法操作。當(dāng)goroutine 試圖去調(diào)用任何原子函數(shù)時(shí)，這些goroutine 都會(huì)自動(dòng)根據(jù)所引用的變量做同步處理。

另外兩個(gè)有用的原子函數(shù)是LoadInt64 和StoreInt64。這兩個(gè)函數(shù)提供了一種安全地讀和寫(xiě)一個(gè)整型值的方式。

package main

import (
    "fmt"
    "sync"
    "sync/atomic"
    "time"
)

var (
    shutdown int64
    wg sync.WaitGroup
)

func main() {
    wg.Add(2)

    go doWork("A")
    go doWork("B")

    time.Sleep(1 * time.Second)

    fmt.Println("Shutdown Now")
    atomic.StoreInt64(&shutdown, 1)
    wg.Wait()
}

func doWork(name string) {
    defer wg.Done()

    for {
        fmt.Printf("Doing %s Work\n", name)
        time.Sleep(250 * time.Millisecond)

        // Do we need to shutdown.
        if atomic.LoadInt64(&shutdown) == 1 {
            fmt.Printf("Shutting %s Down\n", name)
            break
        }
    }
}

主協(xié)程main函數(shù)執(zhí)行了1秒，然后將shutdown中的值設(shè)置為1，A和B的協(xié)程中通過(guò)讀取shutdown是否等于1來(lái)判斷是否結(jié)束協(xié)程。

運(yùn)行結(jié)果：

Doing A Work
Doing B Work
Doing A Work
Doing B Work
Doing A Work
Doing B Work
Doing A Work
Doing B Work
Shutdown Now
Shutting A Down
Shutting B Down

2、互斥鎖

另一種同步訪(fǎng)問(wèn)共享資源的方式是使用互斥鎖（mutex）。互斥鎖這個(gè)名字來(lái)自互斥（mutualexclusion）的概念?；コ怄i用于在代碼上創(chuàng)建一個(gè)臨界區(qū)，保證同一時(shí)間只有一個(gè)goroutine 可以執(zhí)行這個(gè)臨界區(qū)代碼

package main

import (
    "fmt"
    "runtime"
    "sync"
)

var (
    counter int
    wg sync.WaitGroup
    mutex sync.Mutex
)

func main() {
    wg.Add(3)

    go incCounter(1)
    go incCounter(2)
    go incCounter(3)

    wg.Wait()
    fmt.Printf("Final Counter: %d\n", counter)
}

func incCounter(id int) {
    defer wg.Done()

    for count := 0; count < 2; count++ {
        mutex.Lock()
        {
            value := counter
            runtime.Gosched()
            value++
            counter = value
        }
        mutex.Unlock()
    }
}

運(yùn)行結(jié)果：