Go語言學習教程之goroutine和通道的示例詳解

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

這篇文章主要通過A?Tour?of?Go中的例子進行學習，以此了解Go語言中的goroutine和通道，文中的示例代碼講解詳細，感興趣的可以了解一下

goroutine

goroutine是由Go運行時管理的輕量級線程。

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

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

goroutines運行在相同的地址空間中，沒有內(nèi)存隔離，不同的goroutines可以訪問同一個內(nèi)存地址。這樣對共享的內(nèi)存的訪問就可能出現(xiàn)問題，比如有一個全局變量A，goroutine 1開始修改A的數(shù)據(jù)，但是還沒修改完，goroutine 2就開始讀取A的數(shù)據(jù)了，這樣讀到的數(shù)據(jù)可能是不準確的，如果goroutine 2中也要修改A的數(shù)據(jù)，那A的數(shù)據(jù)就處于一種更不確定的狀態(tài)了。所以需要使用互斥鎖，當goroutine 1開始修改A的數(shù)據(jù)之前，先加個鎖，表示這塊內(nèi)存已經(jīng)被鎖上了，等修改完A的數(shù)據(jù)再將鎖解開。在goroutine 1修改數(shù)據(jù)A但還沒修改完的期間，goroutine 2需要修改/讀取A的內(nèi)容，發(fā)現(xiàn)已經(jīng)加鎖，就會進入休眠狀態(tài)，直到變量A的鎖被解開才會執(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的時候，會在主goroutine中執(zhí)行main函數(shù)，當執(zhí)行到go say("a")的時候，會在一個新的goroutine中執(zhí)行say("a")（稱這個子goroutine為goroutine 1），然后主goroutine中繼續(xù)執(zhí)行say("b")，主goroutine和goroutine 1中的函數(shù)執(zhí)行是并發(fā)的。

因為是并發(fā)執(zhí)行，打印出的字符串a(chǎn)和字符串b的順序是無法確定的。

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

通道

通道（channels）是一個類型化的管道（conduit），可以通過<-（通道運算符）來使用通道，對值進行發(fā)送和接收。

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

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

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

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

make(chan int, 100)

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

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

1.無緩存通道例子：

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)
}

如果把以下這兩句注釋掉，運行代碼就會報錯：fatal error: all goroutines are asleep - deadlock!。

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

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

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

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個元素，但是只接收了1個，此時通道中還有2個元素，不會阻塞。

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

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

3.使用通道在goroutines間進行通信的例子：

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)容分為兩部分，在兩個goroutines中分別進行計算，最后再進行求和。

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

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

Range 和 Close

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

執(zhí)行如下語句：

v, ok := <-ch

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

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

注意：

只有發(fā)送方可以關(guān)閉一個通道，接收方不可以。在一個已經(jīng)關(guān)閉的通道上進行發(fā)送會導致一個錯誤（panic）。
通道不像文件，不需要總是關(guān)閉它們。關(guān)閉只有必須告訴接收方不會再來更多值時，才是必須的，比如終止一個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 會一直等待通道關(guān)閉
    close(c)
}

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

如果把close(c) 語句注釋掉，運行代碼，就會報錯：fatal error: all goroutines are asleep - deadlock!。因為for i := range c一直在等通道關(guān)閉，但是整個執(zhí)行過程中并沒有關(guān)閉通道，造成了死鎖。

Select

select語句讓一個goroutine等待多個通信操作。

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

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

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

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

一個select中的default case，在沒有其他case準備好的時候就會運行。

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就會收到一次數(shù)據(jù)，case <-tick中的語句會執(zhí)行，打印一次tick.；500毫秒之后，通道boom會收到數(shù)據(jù)，case <-boom中的語句會執(zhí)行，打印BOOM!，并且使用return結(jié)束程序的執(zhí)行。在這期間，由于for語句是一直在循環(huán)的，當通道tick和通道boom中都沒收到數(shù)據(jù)時，就會執(zhí)行default中的語句：打印一個點并且等待50毫秒。

粗略看了下time.Tick和time.After代碼，兩者返回的值都是類型為<-chan Time的通道，使用輪詢，在滿足時間條件之后，向通道中發(fā)送當前時間。如果想看通道中傳遞的時間數(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

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

Go的標準庫提供了互斥的使用，需要用到sync.Mutex和它的兩個方法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遞增計數(shù)器
func (c *SafeCounter) Inc(key string) {
    c.mu.Lock()
    // 鎖住之后，一次只能有一個goroutine可以訪問映射c.v
    c.v[key]++
    c.mu.Unlock()
}

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

官方留的兩道練習題

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

等價的二叉樹

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

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

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

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

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

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

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

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

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

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

代碼實現(xiàn)

主要部分代碼如下：

package main

import (
    "equbintrees/tree"
    "fmt"
)

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

// 函數(shù)遍歷樹 t，將樹中的所有值依次發(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)
    }
}

// 判斷兩棵樹是否包含相同的值
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，是因為題目要求里面隨機生成的樹的節(jié)點個數(shù)就是10個
            // 一般的樹可以給樹添加一個Len屬性表示節(jié)點個數(shù)，用Len屬性來判斷
            if count == 10 {
                return true
            }
        } else {
            return false
        }
    }
}

網(wǎng)絡爬蟲

使用Go的并發(fā)功能來并發(fā)網(wǎng)絡爬蟲。

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

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

代碼實現(xiàn)

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

但是有問題，因為不能僅憑深度是否達到來判斷是否關(guān)閉通道。給出的例子實際只有4層鏈接，如果設置深度需要到達到5，當遞歸到盡頭的時候就應該關(guān)閉通道了，但是因為沒有達到深度5，沒有關(guān)閉通道，for urls := range ch還會繼續(xù)等通道接收數(shù)據(jù)，但已經(jīng)不會再往通道中發(fā)送數(shù)據(jù)了，造成死鎖?？傊?，手動調(diào)用close(ch)來正確關(guān)閉通道有點難，因為很難找到遞歸和并發(fā)請求時不會再往通道中發(fā)送數(shù)據(jù)的那個時機。

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

主要思路就是使用sync.WaitGroup，用Add方法添加WaitGroup計數(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 // 用于存放表示一個鏈接是否被抓取過的映射
    mux sync.Mutex      // 使用互斥鎖在并發(fā)的執(zhí)行中進行安全的讀寫
}

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

// 檢查鏈接是否已經(jīng)被抓取過
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
}

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

// 遞歸地請求抓去url的數(shù)據(jù)，直到一個最大深度
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)
    }

}