Go語(yǔ)言學(xué)習(xí)教程之goroutine和通道的示例詳解

更新時(shí)間：2022年09月28日 10:22:13 作者：任沫

這篇文章主要通過(guò)A?Tour?of?Go中的例子進(jìn)行學(xué)習(xí)，以此了解Go語(yǔ)言中的goroutine和通道，文中的示例代碼講解詳細(xì)，感興趣的可以了解一下

goroutine

goroutine是由Go運(yùn)行時(shí)管理的輕量級(jí)線程。

go f(x, y, z)在一個(gè)新的goroutine中開(kāi)始執(zhí)行f(x, y,z)。

goroutines運(yùn)行在相同的地址空間中，所以對(duì)共享的內(nèi)存訪問(wèn)必須同步。sync包提供了基本的同步原語(yǔ)（synchronization primitives），比如互斥鎖（mutual exclusion locks）。

goroutines運(yùn)行在相同的地址空間中，沒(méi)有內(nèi)存隔離，不同的goroutines可以訪問(wèn)同一個(gè)內(nèi)存地址。這樣對(duì)共享的內(nèi)存的訪問(wèn)就可能出現(xiàn)問(wèn)題，比如有一個(gè)全局變量A，goroutine 1開(kāi)始修改A的數(shù)據(jù)，但是還沒(méi)修改完，goroutine 2就開(kāi)始讀取A的數(shù)據(jù)了，這樣讀到的數(shù)據(jù)可能是不準(zhǔn)確的，如果goroutine 2中也要修改A的數(shù)據(jù)，那A的數(shù)據(jù)就處于一種更不確定的狀態(tài)了。所以需要使用互斥鎖，當(dāng)goroutine 1開(kāi)始修改A的數(shù)據(jù)之前，先加個(gè)鎖，表示這塊內(nèi)存已經(jīng)被鎖上了，等修改完A的數(shù)據(jù)再將鎖解開(kāi)。在goroutine 1修改數(shù)據(jù)A但還沒(méi)修改完的期間，goroutine 2需要修改/讀取A的內(nèi)容，發(fā)現(xiàn)已經(jīng)加鎖，就會(huì)進(jìn)入休眠狀態(tài)，直到變量A的鎖被解開(kāi)才會(huì)執(zhí)行g(shù)oroutine 2中的修改/讀取。

package main

import (
    "fmt"
    "time"
)

func main() {
    go say("a")
    say("b")
}

func say(s string) {
    for i := 0; i < 5; i++ {
        time.Sleep(2000 * time.Millisecond)
        fmt.Println(s, time.Now().Format("15:04:05.000000"))
    }
}

執(zhí)行go run goroutine.go的時(shí)候，會(huì)在主goroutine中執(zhí)行main函數(shù)，當(dāng)執(zhí)行到go say("a")的時(shí)候，會(huì)在一個(gè)新的goroutine中執(zhí)行say("a")（稱這個(gè)子goroutine為goroutine 1），然后主goroutine中繼續(xù)執(zhí)行say("b")，主goroutine和goroutine 1中的函數(shù)執(zhí)行是并發(fā)的。

因?yàn)槭遣l(fā)執(zhí)行，打印出的字符串a(chǎn)和字符串b的順序是無(wú)法確定的。

（仔細(xì)觀察的話會(huì)發(fā)現(xiàn)打印的前2條數(shù)據(jù)的時(shí)間戳，b的時(shí)間戳在a的后面，但是先打印出了b，這說(shuō)明這次執(zhí)行中，兩者的fmt.Println函數(shù)的執(zhí)行（直到輸出到終端）時(shí)間不同，先拿到了字符串a(chǎn)，但是打印字符串a(chǎn)的fmt.Println執(zhí)行比打印字符串b的函數(shù)執(zhí)行稍稍慢了一點(diǎn)，所以b先出現(xiàn)在了輸出界面上。可能背后還有更復(fù)雜的原因，這里不作深究。）

通道

通道（channels）是一個(gè)類型化的管道（conduit），可以通過(guò)<-（通道運(yùn)算符）來(lái)使用通道，對(duì)值進(jìn)行發(fā)送和接收。

可選的<-操作符指定了通道的方向，如果給出了一個(gè)方向，通道就是定向的，否則就是雙向的。

chan T // 可以被用來(lái)發(fā)送和接收類型為T(mén)的值
chan <- float64 // 只能被用來(lái)發(fā)送float64類型的值
<-chan int // 只能被用來(lái)接收int類型的值

如果有<-操作符的話，數(shù)據(jù)按照箭頭的方向流動(dòng)。

通道在使用前必須被創(chuàng)建：

make(chan int, 100)

通過(guò)內(nèi)置的make函數(shù)創(chuàng)建一個(gè)新的、初始化的通道，接收的參數(shù)是通道類型和一個(gè)可選的容量。容量設(shè)置緩存區(qū)的大小。如果容量是0或者省略了，通道就是非緩存的，只在發(fā)送方和接收方都準(zhǔn)備好的時(shí)候才能通信成功。否則通道就是緩存的，發(fā)送方的緩存區(qū)沒(méi)有滿，或者接收方的緩存區(qū)不為空，就能不阻塞地進(jìn)行通信。

“發(fā)送方的緩存區(qū)沒(méi)有滿，或者接收方的緩存區(qū)不為空，就能不阻塞地進(jìn)行通信。“這句話直白一點(diǎn)說(shuō)，就是如果緩存區(qū)滿了，就不能再往通道中發(fā)送數(shù)據(jù)了（chan <- 數(shù)據(jù) ），如果緩存區(qū)是空的，就不能從通道中接收數(shù)據(jù)了（<-chan）。

1.無(wú)緩存通道例子：

package main

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

var wg sync.WaitGroup

func main() {
    example1()
    wg.Wait() // 等待所有g(shù)oroutines執(zhí)行完成
}

func example1() {
    chan1 := make(chan int)

    wg.Add(1)
    go a(chan1) // 向通道中發(fā)送數(shù)字1、2

    wg.Add(1)
    go b(chan1) // 等待1秒之后，從通道中拿數(shù)據(jù)，拿到的是數(shù)字2

    fmt.Println("接收數(shù)據(jù)A", <-chan1) // 這里拿到的是數(shù)字1
}

func a(chan1 chan int) {
    defer wg.Done()
    chan1 <- 1
    chan1 <- 2
}

func b(chan1 chan int) {
    defer wg.Done()
    time.Sleep(time.Second)
    fmt.Println("接收數(shù)據(jù)B", <-chan1)
}

如果把以下這兩句注釋掉，運(yùn)行代碼就會(huì)報(bào)錯(cuò)：fatal error: all goroutines are asleep - deadlock!。

    wg.Add(1)
    go b(chan1) // 等待1秒之后，從通道中拿數(shù)據(jù)

把這句注釋掉，代碼變成了往無(wú)緩存通道中發(fā)送了2個(gè)元素，但是只接收了1個(gè)元素。由于向通道中發(fā)送的元素2沒(méi)被接收，通道會(huì)阻塞，sync包又在等待數(shù)字2的發(fā)送（chan1 <- 2）完成，就造成了死鎖。

最終在無(wú)緩存通道中的元素個(gè)數(shù)為0，無(wú)緩存通道就不會(huì)阻塞。

2.有緩存通道例子：

...
var wg sync.WaitGroup

func main() {
    example2()
    wg.Wait() // 等待所有g(shù)oroutines執(zhí)行完成
}

func example2() {
    chan1 := make(chan int, 2)

    wg.Add(1)
    go a(chan1) // 向通道中發(fā)送數(shù)字1、2、3

    fmt.Println("接收數(shù)據(jù)", <-chan1)
}

func a(chan1 chan int) {
    defer wg.Done()
    chan1 <- 1
    chan1 <- 2
    chan1 <- 3
}

func b(chan1 chan int) {
    defer wg.Done()
    time.Sleep(time.Second)
    fmt.Println("接收數(shù)據(jù)", <-chan1)
}

以上代碼向容量為2的緩存通道中發(fā)送了3個(gè)元素，但是只接收了1個(gè)，此時(shí)通道中還有2個(gè)元素，不會(huì)阻塞。

如果在a函數(shù)的最后一行再加上一句chan1 <- 4，再執(zhí)行代碼，就會(huì)報(bào)錯(cuò)fatal error: all goroutines are asleep - deadlock!。因?yàn)榘l(fā)送了4個(gè)元素，只接收了1個(gè)元素，還剩3個(gè)元素沒(méi)被接收，3 > 2，緩存已經(jīng)滿了，由于代碼中沒(méi)有別的地方來(lái)接收元素，通道阻塞，但是sync包又在等待chan1 <- 4的完成，所以會(huì)造成死鎖。

最終在有緩存通道中的元素個(gè)數(shù)小于等于容量，有緩存通道就不會(huì)阻塞。

3.使用通道在goroutines間進(jìn)行通信的例子：

func main() {
    example3()
}

func example3() {
    s := []int{7, 2, 8, -9, 4, 0}

    c := make(chan int)
    go sum(s[:len(s)/2], c)
    go sum(s[len(s)/2:], c)
    x, y := <-c, <-c

    fmt.Println(x, y, x+y)
}

func sum(s []int, c chan int) {
    sum := 0
    for _, v := range s {
        sum += v
    }
    c <- sum
}

這段代碼將數(shù)組的內(nèi)容分為兩部分，在兩個(gè)goroutines中分別進(jìn)行計(jì)算，最后再進(jìn)行求和。

這里兩個(gè)子goroutines是與主goroutine并發(fā)執(zhí)行的，但主goroutine中的x, y := <-c, <-c依然拿到了兩個(gè)子goroutines中往通道發(fā)送的數(shù)據(jù)（c <- sum）。這是因?yàn)橥ǖ赖陌l(fā)送和接收會(huì)阻塞，直到另一邊準(zhǔn)備好。

x拿到的是先計(jì)算完的和，y拿到的是后計(jì)算完的和，x，y的值是不確定的，可能是-5 17 或者 17 -5，就看哪個(gè)子goroutine中的計(jì)算先完成。

Range 和 Close

發(fā)送方可以close一個(gè)通道來(lái)表明沒(méi)有更多的值會(huì)被發(fā)送。接收方可以通過(guò)賦值第二個(gè)參數(shù)給接收表達(dá)式，測(cè)試一個(gè)通道是否已經(jīng)被關(guān)閉。

執(zhí)行如下語(yǔ)句：

v, ok := <-ch

如果沒(méi)有更多的值要接收，并且通道已經(jīng)關(guān)閉了，ok的值就為false。

for i := range c循環(huán)，從通道中重復(fù)地接收值，直到通道關(guān)閉。

注意：

只有發(fā)送方可以關(guān)閉一個(gè)通道，接收方不可以。在一個(gè)已經(jīng)關(guān)閉的通道上進(jìn)行發(fā)送會(huì)導(dǎo)致一個(gè)錯(cuò)誤（panic）。
通道不像文件，不需要總是關(guān)閉它們。關(guān)閉只有必須告訴接收方不會(huì)再來(lái)更多值時(shí)，才是必須的，比如終止一個(gè)range循環(huán)。

func main() {
    c := make(chan int, 10)
    go fibonacci(cap(c), c)
    for i := range c {
        fmt.Println(i)
    }
}

func fibonacci(n int, c chan int) {
    x, y := 0, 1
    for i := 0; i < n; i++ {
        c <- x
        x, y = y, x+y
    }
    // 必須在遍歷結(jié)束之后關(guān)閉通道
    // 否則 for i := range c 會(huì)一直等待通道關(guān)閉
    close(c)
}

以上代碼求斐波那契數(shù)列，依次將求得的值發(fā)送到通道。

如果把close(c) 語(yǔ)句注釋掉，運(yùn)行代碼，就會(huì)報(bào)錯(cuò)：fatal error: all goroutines are asleep - deadlock!。因?yàn)?code>for i := range c一直在等通道關(guān)閉，但是整個(gè)執(zhí)行過(guò)程中并沒(méi)有關(guān)閉通道，造成了死鎖。

Select

select語(yǔ)句讓一個(gè)goroutine等待多個(gè)通信操作。

一個(gè)select 會(huì)阻塞，直到它的cases中的一個(gè)可以運(yùn)行，然后它就會(huì)執(zhí)行該case。如果多個(gè)通信都準(zhǔn)備好了，就會(huì)隨機(jī)選擇一個(gè)。

func main() {
    c := make(chan int)
    quit := make(chan int)
    go func() {
        for i := 0; i < 10; i++ {
            fmt.Println(<-c)
        }
        quit <- 0
    }()
    fibonacci(c, quit)
}

func fibonacci(c, quit chan int) {
    x, y := 0, 1
    for {
        select {
        case c <- x:
            x, y = y, x+y
        case <-quit:
            fmt.Println("quit")
            return
        }
    }
}

上述代碼還是實(shí)現(xiàn)一個(gè)斐波那契數(shù)列的計(jì)算。

在子goroutine（稱之為goroutine 1）中循環(huán)10次，依次從通道c中接收數(shù)據(jù)，循環(huán)結(jié)束之后，將數(shù)字0發(fā)送到通道quit。

在主goroutine中，調(diào)用fibonacci函數(shù)：

c <- x是向通道中發(fā)送數(shù)據(jù)，只要有地方從通道中接收數(shù)據(jù)，向通道中發(fā)送數(shù)據(jù)就能繼續(xù)運(yùn)行。每次在goroutine 1的循環(huán)中<-c，主goroutine中的select語(yǔ)句中的case c <- x中的語(yǔ)句就會(huì)執(zhí)行。
<-quit是從通道中接收數(shù)據(jù)，只要有地方向通道中發(fā)送數(shù)據(jù)，從通道中接收數(shù)據(jù)就能繼續(xù)運(yùn)行。當(dāng)goroutine 1中循環(huán)結(jié)束之后quit <- 0，case <-quit中的語(yǔ)句就會(huì)執(zhí)行。

一個(gè)select中的default case，在沒(méi)有其他case準(zhǔn)備好的時(shí)候就會(huì)運(yùn)行。

package main

import (
    "fmt"
    "time"
)

func main() {
    tick := time.Tick(100 * time.Millisecond)
    boom := time.After(500 * time.Millisecond)

    for {
        select {
        case <-tick:
            fmt.Println("tick.")
        case <-boom:
            fmt.Println("BOOM!")
            return
        default:
            fmt.Println("    .")
            time.Sleep(50 * time.Millisecond)
        }
    }
}

每隔100毫秒，通道tick就會(huì)收到一次數(shù)據(jù)，case <-tick中的語(yǔ)句會(huì)執(zhí)行，打印一次tick.；500毫秒之后，通道boom會(huì)收到數(shù)據(jù)，case <-boom中的語(yǔ)句會(huì)執(zhí)行，打印BOOM!，并且使用return結(jié)束程序的執(zhí)行。在這期間，由于for語(yǔ)句是一直在循環(huán)的，當(dāng)通道tick和通道boom中都沒(méi)收到數(shù)據(jù)時(shí)，就會(huì)執(zhí)行default中的語(yǔ)句：打印一個(gè)點(diǎn)并且等待50毫秒。

粗略看了下time.Tick和time.After代碼，兩者返回的值都是類型為<-chan Time的通道，使用輪詢，在滿足時(shí)間條件之后，向通道中發(fā)送當(dāng)前時(shí)間。如果想看通道中傳遞的時(shí)間數(shù)據(jù)的話，可以使用以下代碼：

package main

import (
    "fmt"
    "time"
)

func main() {
    tick := time.Tick(100 * time.Millisecond)
    boom := time.After(500 * time.Millisecond)
    var x, y time.Time
    for {
        select {
        case x, _ = <-tick:
            fmt.Println(x, "tick.")
        case y, _ = <-boom:
            fmt.Println(y, "BOOM!")
            return
        default:
            fmt.Println("    .")
            time.Sleep(50 * time.Millisecond)
        }
    }
}

sync.Mutex

如果我們想要避免沖突，確保一次只有一個(gè)goroutine可以訪問(wèn)一個(gè)變量（這個(gè)概念稱為互斥），則可以使用互斥鎖（mutex）。

Go的標(biāo)準(zhǔn)庫(kù)提供了互斥的使用，需要用到sync.Mutex和它的兩個(gè)方法Lock和Unlock。

package main

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

func main() {
    c := SafeCounter{v: make(map[string]int)}
    for i := 0; i < 1000; i++ {
        go c.Inc("somekey")
    }
    time.Sleep(time.Second)
    fmt.Println(c.Value("somekey"))
}


type SafeCounter struct {
    mu sync.Mutex
    v  map[string]int
}

// 使用給定的key遞增計(jì)數(shù)器
func (c *SafeCounter) Inc(key string) {
    c.mu.Lock()
    // 鎖住之后，一次只能有一個(gè)goroutine可以訪問(wèn)映射c.v
    c.v[key]++
    c.mu.Unlock()
}

// 返回 給定key的 計(jì)數(shù)器的當(dāng)前值
func (c *SafeCounter) Value(key string) int {
    c.mu.Lock()
    // 鎖住之后，一次只能有一個(gè)goroutine可以訪問(wèn)映射c.v
    defer c.mu.Unlock()
    return c.v[key]
}

官方留的兩道練習(xí)題

官方留了兩道練習(xí)題，沒(méi)有給出完整的代碼?？梢宰鳛榱私饬艘陨现R(shí)之后的練手。

等價(jià)的二叉樹(shù)

有很多不同的二叉樹(shù)，存儲(chǔ)著相同的值的序列。例如，下圖兩棵二叉樹(shù)存儲(chǔ)的序列是1, 1, 2, 3, 5, 8, 13。

1.實(shí)現(xiàn)Walk函數(shù)。

2.測(cè)試Walk函數(shù)。

函數(shù)tree.New(k)構(gòu)造了一個(gè)隨機(jī)結(jié)構(gòu)（但總是排序的）的二叉樹(shù)來(lái)存儲(chǔ)值k，2k，3k，...，10k。

創(chuàng)建一個(gè)新的通道ch并開(kāi)始遍歷：

go Walk(tree.New(1), ch)

然后打印樹(shù)中包含的10個(gè)值，應(yīng)該是數(shù)字1，2，3，...，10。

3.實(shí)現(xiàn)Same函數(shù)，使用Walk來(lái)決定t1和t2是否存儲(chǔ)相同的值。

4.測(cè)試Same函數(shù)：

Same(tree.New(1), tree.New(1)) 應(yīng)該返回true，
Same(tree.New(1), tree.New(2)) 應(yīng)該返回false。

代碼實(shí)現(xiàn)

主要部分代碼如下：

package main

import (
    "equbintrees/tree"
    "fmt"
)

func main() {
    tree1 := tree.New(1)
    tree2 := tree.New(2)
    fmt.Println(Same(tree1, tree2))
}

// 函數(shù)遍歷樹(shù) t，將樹(shù)中的所有值依次發(fā)送到通道中
func Walk(t *tree.Tree, ch chan int) {
    if t == nil {
        return
    }
    if t.Left != nil {
        Walk(t.Left, ch)
    }
    ch <- t.Value
    if t.Right != nil {
        Walk(t.Right, ch)
    }
}

// 判斷兩棵樹(shù)是否包含相同的值
func Same(t1, t2 *tree.Tree) bool {
    ch1 := make(chan int)
    ch2 := make(chan int)

    go Walk(t1, ch1)
    go Walk(t2, ch2)

    var count int

    for {
        if <-ch1 == <-ch2 {
            count++
            // 這里的count等于10，是因?yàn)轭}目要求里面隨機(jī)生成的樹(shù)的節(jié)點(diǎn)個(gè)數(shù)就是10個(gè)
            // 一般的樹(shù)可以給樹(shù)添加一個(gè)Len屬性表示節(jié)點(diǎn)個(gè)數(shù)，用Len屬性來(lái)判斷
            if count == 10 {
                return true
            }
        } else {
            return false
        }
    }
}

網(wǎng)絡(luò)爬蟲(chóng)

使用Go的并發(fā)功能來(lái)并發(fā)網(wǎng)絡(luò)爬蟲(chóng)。

修改Crawl函數(shù)來(lái)并發(fā)獲取URLs，并且相同的URL不會(huì)獲取2次。

提示：你可以使用映射緩存已經(jīng)獲取到的URL，但是只使用映射對(duì)于并發(fā)使用來(lái)說(shuō)是不安全的。

代碼實(shí)現(xiàn)

這部分我嘗試實(shí)現(xiàn)了下，主要思路是在遞歸的過(guò)程中，將遍歷到鏈接中包含的urls發(fā)送到通道ch中，用for urls := range ch遍歷通道中的元素，以此來(lái)等待所有發(fā)送到通道中的urls都被接收，在遞歸過(guò)程中判斷深度是否達(dá)到4，達(dá)到4之后調(diào)用close(ch)關(guān)閉通道。

但是有問(wèn)題，因?yàn)椴荒軆H憑深度是否達(dá)到來(lái)判斷是否關(guān)閉通道。給出的例子實(shí)際只有4層鏈接，如果設(shè)置深度需要到達(dá)到5，當(dāng)遞歸到盡頭的時(shí)候就應(yīng)該關(guān)閉通道了，但是因?yàn)闆](méi)有達(dá)到深度5，沒(méi)有關(guān)閉通道，for urls := range ch還會(huì)繼續(xù)等通道接收數(shù)據(jù)，但已經(jīng)不會(huì)再往通道中發(fā)送數(shù)據(jù)了，造成死鎖。總之，手動(dòng)調(diào)用close(ch)來(lái)正確關(guān)閉通道有點(diǎn)難，因?yàn)楹茈y找到遞歸和并發(fā)請(qǐng)求時(shí)不會(huì)再往通道中發(fā)送數(shù)據(jù)的那個(gè)時(shí)機(jī)。

我從這個(gè)鏈接找到了大佬的代碼實(shí)現(xiàn)：https://rmoff.net/2020/07/03/learning-golang-some-rough-notes-s01e10-concurrency-web-crawler/

主要思路就是使用sync.WaitGroup，用Add方法添加WaitGroup計(jì)數(shù)，用wg.Wait()等待所有的goroutines執(zhí)行結(jié)束。

主要部分代碼如下：

func main() {
    wg := &sync.WaitGroup{}
    wg.Add(1)
    go Crawl("https://golang.org/", 5, fetcher, wg)
    wg.Wait()
}

type URLs struct {
    c   map[string]bool // 用于存放表示一個(gè)鏈接是否被抓取過(guò)的映射
    mux sync.Mutex      // 使用互斥鎖在并發(fā)的執(zhí)行中進(jìn)行安全的讀寫(xiě)
}

var u URLs = URLs{c: make(map[string]bool)}

// 檢查鏈接是否已經(jīng)被抓取過(guò)
func (u URLs) IsCrawled(url string) bool {
    fmt.Printf("\n?? Checking if %v has been crawled…", url)
    u.mux.Lock()
    defer u.mux.Unlock()
    if _, ok := u.c[url]; ok == false {
        fmt.Printf("…it hasn't\t")
        return false
    }
    fmt.Printf("…it has\t")
    return true
}

// 將鏈接標(biāo)記為抓取過(guò)
func (u URLs) Crawled(url string) {
    u.mux.Lock()
    u.c[url] = true
    u.mux.Unlock()
}

// 遞歸地請(qǐng)求抓去url的數(shù)據(jù)，直到一個(gè)最大深度
func Crawl(url string, depth int, fetcher Fetcher, wg *sync.WaitGroup) {
    defer wg.Done()

    if depth <= 0 {
        return
    }

    if u.IsCrawled(url) == true {
        return
    }

    fmt.Printf("\n?? Crawling %v", url)
    body, urls, err := fetcher.Fetch(url)
    u.Crawled(url)

    if err != nil {
        fmt.Println(err)
        return
    }
    fmt.Printf("\n\t->? found: %s %q\n", url, body)

    for _, z := range urls {
        wg.Add(1)

        go Crawl(z, depth-1, fetcher, wg)
    }

}