[[232153]]

Go 是一個內置支持并發編程的語言。借助使用 go 關鍵字去創建協程goroutine（輕量級線程）和在 Go 中提供的 使用 信道 和 其它的并發 同步方法，使得并發編程變得很容易、很靈活和很有趣。

另一方面，Go 并不會阻止一些因 Go 程序員粗心大意或者缺乏經驗而造成的并發編程錯誤。在本文的下面部分將展示一些在 Go 編程中常見的并發編程錯誤，以幫助 Go 程序員們避免再犯類似的錯誤。

需要同步的時候沒有同步

代碼行或許 不是按出現的順序運行的。

在下面的程序中有兩個錯誤。

• ***，在 main 協程中讀取 b 和在新的 協程 中寫入 b 可能導致數據爭用。
• 第二，條件 b == true 并不能保證在 main 協程 中的 a != nil。在新的協程中編譯器和 CPU 可能會通過 重排序指令 進行優化，因此，在運行時 b 賦值可能發生在 a 賦值之前，在 main 協程 中當 a 被修改后，它將會讓部分 a 一直保持為 nil。

package main
 
import (
    "time"
    "runtime"
)
 
func main() {
    var a []int // nil
    var b bool  // false
 
    // a new goroutine
    go func () {
        a = make([]int, 3)
        b = true // write b
    }()
 
    for !b { // read b
        time.Sleep(time.Second)
        runtime.Gosched()
    }
    a[0], a[1], a[2] = 0, 1, 2 // might panic
}

上面的程序或者在一臺計算機上運行的很好，但是在另一臺上可能會引發異常。或者它可能運行了 N 次都很好，但是可能在第 (N+1) 次引發了異常。

我們將使用 sync 標準包中提供的信道或者同步方法去確保內存中的順序。例如，

package main
 
func main() {
    var a []int = nil
    c := make(chan struct{})
 
    // a new goroutine
    go func () {
        a = make([]int, 3)
        c <- struct{}{}
    }()
 
    <-c
    a[0], a[1], a[2] = 0, 1, 2
}
 

使用 time.Sleep 調用去做同步

我們先來看一個簡單的例子。

package main
 
import (
    "fmt"
    "time"
)
 
func main() {
    var x = 123
 
    go func() {
        x = 789 // write x
    }()
 
    time.Sleep(time.Second)
    fmt.Println(x) // read x
}
 

我們預期程序將打印出 789。如果我們運行它，通常情況下，它確定打印的是 789。但是，這個程序使用的同步方式好嗎？No！原因是 Go 運行時并不保證 x 的寫入一定會發生在 x 的讀取之前。在某些條件下，比如在同一個操作系統上，大部分 CPU 資源被其它運行的程序所占用的情況下，寫入 x 可能就會發生在讀取 x 之后。這就是為什么我們在正式的項目中，從來不使用 time.Sleep 調用去實現同步的原因。

我們來看一下另外一個示例。

package main
 
import (
    "fmt"
    "time"
)
 
var x = 0
 
func main() {
    var num = 123
    var p = &num
 
    c := make(chan int)
 
    go func() {
        c <- *p + x
    }()
 
    time.Sleep(time.Second)
    num = 789
    fmt.Println(<-c)
}

你認為程序的預期輸出是什么？123 還是 789？事實上它的輸出與編譯器有關。對于標準的 Go 編譯器 1.10 來說，這個程序很有可能輸出是 123。但是在理論上，它可能輸出的是 789，或者其它的隨機數。

現在，我們來改變 c <- *p + x 為 c <- *p，然后再次運行這個程序。你將會發現輸出變成了 789 （使用標準的 Go 編譯器 1.10）。這再次說明它的輸出是與編譯器相關的。

是的，在上面的程序中存在數據爭用。表達式 *p 可能會被先計算、后計算、或者在處理賦值語句 num = 789 時計算。time.Sleep 調用并不能保證 *p 發生在賦值語句處理之前進行。

對于這個特定的示例，我們將在新的協程創建之前，將值保存到一個臨時值中，然后在新的協程中使用臨時值去消除數據爭用。

...
    tmp := *p + x
    go func() {
        c <- tmp
    }()
...

使協程掛起

掛起協程是指讓協程一直處于阻塞狀態。導致協程被掛起的原因很多。比如，

• 一個協程嘗試從一個 nil 信道中或者從一個沒有其它協程給它發送值的信道中檢索數據。
• 一個協程嘗試去發送一個值到 nil 信道，或者發送到一個沒有其它的協程接收值的信道中。
• 一個協程被它自己死鎖。
• 一組協程彼此死鎖。
• 當運行一個沒有 default 分支的 select 代碼塊時，一個協程被阻塞，以及在 select 代碼塊中  case 關鍵字后的所有信道操作保持阻塞狀態。

除了有時我們為了避免程序退出，特意讓一個程序中的 main 協程保持掛起之外，大多數其它的協程掛起都是意外情況。Go 運行時很難判斷一個協程到底是處于掛起狀態還是臨時阻塞。因此，Go 運行時并不會去釋放一個掛起的協程所占用的資源。

在 誰先響應誰獲勝 的信道使用案例中，如果使用的 future 信道容量不夠大，當嘗試向 Future 信道發送結果時，一些響應較慢的信道將被掛起。比如，如果調用下面的函數，將有 4 個協程處于永遠阻塞狀態。

func request() int {
    c := make(chan int)
    for i := 0; i < 5; i++ {
        i := i
        go func() {
            c <- i // 4 goroutines will hang here.
        }()
    }
    return <-c
}

為避免這 4 個協程一直處于掛起狀態， c 信道的容量必須至少是  4。

在 實現誰先響應誰獲勝的第二種方法 的信道使用案例中，如果將 future 信道用做非緩沖信道，那么有可能這個信息將永遠也不會有響應而掛起。例如，如果在一個協程中調用下面的函數，協程可能會掛起。原因是，如果接收操作  <-c 準備就緒之前，五個發送操作全部嘗試發送，那么所有的嘗試發送的操作將全部失敗，因此那個調用者協程將永遠也不會接收到值。

func request() int {
    c := make(chan int)
    for i := 0; i < 5; i++ {
        i := i
        go func() {
            select {
            case c <- i:
            default:
            }
        }()
    }
    return <-c
}

將信道 c 變成緩沖信道將保證五個發送操作中的至少一個操作會發送成功，這樣，上面函數中的那個調用者協程將不會被掛起。

在 sync 標準包中拷貝類型值

在實踐中，sync 標準包中的類型值不會被拷貝。我們應該只拷貝這個值的指針。

下面是一個錯誤的并發編程示例。在這個示例中，當調用 Counter.Value 方法時，將拷貝一個 Counter 接收值。作為接收值的一個字段，Counter 接收值的各個 Mutex 字段也會被拷貝。拷貝不是同步發生的，因此，拷貝的 Mutex 值可能會出錯。即便是沒有錯誤，拷貝的 Counter 接收值的訪問保護也是沒有意義的。

import "sync"
 
type Counter struct {
    sync.Mutex
    n int64
}
 
// This method is okay.
func (c *Counter) Increase(d int64) (r int64) {
    c.Lock()
    c.n += d
    r = c.n
    c.Unlock()
    return
}
 
// The method is bad. When it is called, a Counter
// receiver value will be copied.
func (c Counter) Value() (r int64) {
    c.Lock()
    r = c.n
    c.Unlock()
    return
}
 

我們只需要改變 Value 接收類型方法為指針類型 *Counter，就可以避免拷貝 Mutex 值。

在官方的 Go SDK 中提供的 go vet 命令將會報告潛在的錯誤值拷貝。

在錯誤的地方調用 sync.WaitGroup 的方法

每個 sync.WaitGroup 值維護一個內部計數器，這個計數器的初始值為 0。如果一個 WaitGroup 計數器的值是 0，調用 WaitGroup 值的 Wait 方法就不會被阻塞，否則，在計數器值為 0 之前，這個調用會一直被阻塞。

為了讓 WaitGroup 值的使用有意義，當一個 WaitGroup 計數器值為 0 時，必須在相應的 WaitGroup 值的  Wait 方法調用之前，去調用 WaitGroup 值的 Add 方法。

例如，下面的程序中，在不正確位置調用了 Add 方法，這將使***打印出的數字不總是 100。事實上，這個程序***打印的數字可能是在 [0, 100) 范圍內的一個隨意數字。原因就是 Add 方法的調用并不保證一定會發生在 Wait 方法調用之前。

package main
 
import (
    "fmt"
    "sync"
    "sync/atomic"
)
 
func main() {
    var wg sync.WaitGroup
    var x int32 = 0
    for i := 0; i < 100; i++ {
        go func() {
            wg.Add(1)
            atomic.AddInt32(&x, 1)
            wg.Done()
        }()
    }
 
    fmt.Println("To wait ...")
    wg.Wait()
    fmt.Println(atomic.LoadInt32(&x))
}
 

為讓程序的表現符合預期，在 for 循環中，我們將把 Add 方法的調用移動到創建的新協程的范圍之外，修改后的代碼如下。

...
    for i := 0; i < 100; i++ {
        wg.Add(1)
        go func() {
            atomic.AddInt32(&x, 1)
            wg.Done()
        }()
    }
...

不正確使用 futures 信道

在 信道使用案例 的文章中，我們知道一些函數將返回 futures 信道。假設 fa 和 fb 就是這樣的兩個函數，那么下面的調用就使用了不正確的 future 參數。

doSomethingWithFutureArguments(<-fa(), <-fb())

在上面的代碼行中，兩個信道接收操作是順序進行的，而不是并發的。我們做如下修改使它變成并發操作。

ca, cb := fa(), fb()
doSomethingWithFutureArguments(<-c1, <-c2)

沒有等協程的***的活動的發送結束就關閉信道

Go 程序員經常犯的一個錯誤是，還有一些其它的協程可能會發送值到以前的信道時，這個信道就已經被關閉了。當這樣的發送（發送到一個已經關閉的信道）真實發生時，將引發一個異常。

這種錯誤在一些以往的著名 Go 項目中也有發生，比如在 Kubernetes 項目中的 這個 bug 和 這個 bug。

如何安全和優雅地關閉信道，請閱讀 這篇文章。

在值上做 64 位原子操作時沒有保證值地址 64 位對齊

到目前為止（Go 1.10），在標準的 Go 編譯器中，在一個 64 位原子操作中涉及到的值的地址要求必須是 64 位對齊的。如果沒有對齊則導致當前的協程異常。對于標準的 Go 編譯器來說，這種失敗僅發生在 32 位的架構上。請閱讀 內存布局 去了解如何在一個 32 位操作系統上保證 64 位對齊。

沒有注意到大量的資源被 time.After 函數調用占用

在 time 標準包中的 After 函數返回 一個延遲通知的信道。這個函數在某些情況下用起來很便捷，但是，每次調用它將創建一個 time.Timer 類型的新值。這個新創建的 Timer 值在通過傳遞參數到  After 函數指定期間保持激活狀態，如果在這個期間過多的調用了該函數，可能會有太多的 Timer 值保持激活，這將占用大量的內存和計算資源。

例如，如果調用了下列的 longRunning 函數，將在一分鐘內產生大量的消息，然后在某些周期內將有大量的 Timer 值保持激活，即便是大量的這些 Timer 值已經沒用了也是如此。

import (
    "fmt"
    "time"
)
 
// The function will return if a message arrival interval
// is larger than one minute.
func longRunning(messages <-chan string) {
    for {
        select {
        case <-time.After(time.Minute):
            return
        case msg := <-messages:
            fmt.Println(msg)
        }
    }
}

為避免在上述代碼中創建過多的 Timer 值，我們將使用一個單一的 Timer 值去完成同樣的任務。

func longRunning(messages <-chan string) {
    timer := time.NewTimer(time.Minute)
    defer timer.Stop()
 
    for {
        select {
        case <-timer.C:
            return
        case msg := <-messages:
            fmt.Println(msg)
            if !timer.Stop() {
                <-timer.C
            }
        }
 
        // The above "if" block can also be put here.
 
        timer.Reset(time.Minute)
    }
}
 

不正確地使用 time.Timer 值

在***，我們將展示一個符合語言使用習慣的 time.Timer 值的使用示例。需要注意的一個細節是，那個 Reset 方法總是在停止或者 time.Timer 值釋放時被使用。

在 select 塊的***個 case 分支的結束部分，time.Timer 值被釋放，因此，我們不需要去停止它。但是必須在第二個分支中停止定時器。如果在第二個分支中 if 代碼塊缺失，它可能至少在 Reset 方法調用時，會（通過 Go 運行時）發送到 timer.C 信道，并且那個 longRunning 函數可能會早于預期返回，對于 Reset 方法來說，它可能僅僅是重置內部定時器為 0，它將不會清理（耗盡）那個發送到 timer.C 信道的值。

例如，下面的程序很有可能在一秒內而不是十秒時退出。并且更重要的是，這個程序并不是 DRF 的（LCTT 譯注：data race free，多線程程序的一種同步程度）。

package main
 
import (
    "fmt"
    "time"
)
 
func main() {
    start := time.Now()
    timer := time.NewTimer(time.Second/2)
    select {
    case <-timer.C:
    default:
        time.Sleep(time.Second) // go here
    }
    timer.Reset(time.Second * 10)
    <-timer.C
    fmt.Println(time.Since(start)) // 1.000188181s
}

當 time.Timer 的值不再被其它任何一個東西使用時，它的值可能被停留在一種非停止狀態，但是，建議在結束時停止它。

在多個協程中如果不按建議使用 time.Timer 值并發，可能會有 bug 隱患。

我們不應該依賴一個 Reset 方法調用的返回值。Reset 方法返回值的存在僅僅是為了兼容性目的。