Go36-31-sync.WaitGroup和sync.Once

sync.WaitGroup

之前在協調多個goroutine的時候，使用了通道。基本都是按下面這樣來使用的：

package main

import "fmt"

func main() {
    done := make(chan struct{})
    count := 5

    for i := 0; i < count; i++ {
        go func(i int) {
            defer func() {
                done <- struct{}{}
            }()
            fmt.Println(i)
        }(i)
    }

    for j := 0; j < count; j++ {
        <- done
    }
    fmt.Println("Over")
}

這里有一個問題，要保證主goroutine最后從通道接收元素的的次數需要與之前其他goroutine發送元素的次數相同。
其實，在這種應用場景下，可以選用另外一個同步工具，就是這里要講的sync包的WaitGroup類型。

使用方法

sync.WaitGroup類型，它比通道更加適合實現這種一對多的goroutine協作流程。WaitGroup是開箱即用的，也是并發安全的。同時，與之前提到的同步工具一樣，它一旦被真正的使用就不能被復制了。
WaitGroup擁有三個指針方法，可以想象該類型中有一個計數器，默認值是0，下面的方法就是操作或判斷計數器：

• Add ： 增加或減少計數器的值。一般情況下，會用這個方法來記錄需要等待的goroutine的數量
• Done ： 用于對其所屬值中計數器的值進行減一操作，就是Add(-1)，可以在defer語句中調用它
• Wait ： 阻塞當前的goroutine，直到所屬值中的計數器歸零。

現在就用WaitGroup來改造開篇的程序：

package main

import (
    "fmt"
    "sync"
)

func main() {
    var wg sync.WaitGroup  // 開箱即用，所以直接聲明就好了，沒必要用短變量聲明
    // wg := sync.WaitGroup{}  // 短變量聲明可以這么寫
    count := 5

    for i := 0; i < count; i++ {
        wg.Add(1)
        go func(i int) {
            defer wg.Done()
            fmt.Println(i)
        }(i)
    }

    wg.Wait()
    fmt.Println("Over")
}

改造后，在主goroutine最后等待退出的部分現在看著要美觀多了。這個就是WaitGroup典型的應用場景了。

注意的事項

計數器不能小于0
在sync.WaitGroup類型值中計數器的值是不可以小于0的。一旦小于0會引發panic，不適當的調用Done方法和Add方法就有可能使它小于0而引發panic。

盡早增加計數器的值
如果在對它的Add方法的首次調用，與對它的Wait方法的調用是同時發起的。比如，在同時啟動的兩個goroutine中，分別調用這兩個方法，那就就有可能會讓這里的Add方法拋出一個panic。并且這種情況不太容易，應該予以重視。所以雖然WaitGroup值本身并不需要初始化，但是盡早的增加其計數器的值是非要必要的。

復用的情況
WaitGroup的值是可以被復用的，但需要保證其計數周期的完整性。這里的計數周期指的是這樣一個過程：該值中的計數器值由0變為了某個正整數，而后又經過一系列的變化，最終由某個正整數又變回了0。這個過程可以被視為一個計數周期。在一個此類的生命周期中，它可以經歷任意多個計數周期。但是，只有在它走完當前的計數周期后，才能夠開始下一個計數周期。
也就是說，如果一個此類值的Wait方法在它的某個計數周期中被調用，那么就會立即阻塞當前的goroutine，直至這個計數周期完成。在這種情況下，該值的下一個計數周期必須要等到這個Wait方法執行結束之后，才能夠開始。
Wait方法是有一個執行的過程的，如果在這個方法執行期間，跨越了兩個計數周期，就會引發一個panic。比如，當前的goroutine調用了Wait方法而阻塞了。另一個goroutine調用了Done方法使計數器變成了0。此時會喚醒之前阻塞的goroutine，并且去執行Wait方法中其余的代碼（這里還在這行Wait方法，執行的是源碼sync.Wait方法里的代碼，不是我們自己寫的程序的Wait之后的代碼）。在這個時候，又有一個goroutine調用了Add方法，使計數器的值又從0變為了某個正整數。此時正在執行的Wait方法就會立即拋出一個panic。

小結

上面給了3種會引發panic的情況。關于后兩種情況，建議如下：

不要把增加計數器值的操作和調用Wait方法的代碼，放在不同的goroutine中執行。
就是要杜絕對同一個WatiGroup值的兩種操作的并發執行。

后面提到的兩種情況，不是每次都會發生，通常需要反復的實驗才能夠引發panic的情況。雖然不是每次都發生，但是在長期運行的過程中，這種情況是必然會出現的，應該予以重視并且避免。
如果對復現這些異常情況感興趣，可以看一下sync代碼包中的waitgroup_test.go文件。其中的名稱以TestWaitGroupMisuse為前綴的測試函數，很好的展示了這些異常情況發生的條件。

sync.Once

與sync.WaitGroup類型一樣，Sync.Once類型也屬于結構體類型，同樣也是開箱即用和并發安全的。由于這個類型中包含了一個sync.Mutex類型的字段，所以復制改類型的值也會導致功能失效。

使用方法

Do方法
Once類型的Do方法只接收一個參數，參數的類型必須是func()，即無參數無返回的函數。該方法的功能并不是對每一種參數函數都只執行一次，而是只執行首次被調用時傳入的那個函數，并且之后不會再執行任何參數函數。所以，如果有多個需要執行一次的函數，應該為它們每一個都分配一個sync.Once類型的值。
基本用法如下：

package main

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

func main() {
    var counter uint32
    var once sync.Once
    once.Do(func() {
        atomic.AddUint32(&counter, 1)
    })
    fmt.Println("counter:", counter)
    // 這次調用不會被執行
    once.Do(func() {
        atomic.AddUint32(&counter, 2)
    })
    fmt.Println("counter:", counter)
}

done字段
Once類型中還要一個名為done的uint32類型的字段。它的作用是記錄所屬值的Do方法被調用的次數。不過改字段的值只可能是0或1.一旦Do方法的首次調用完成，它的值就會從0變為1。
關于done的類型，其實用布爾類型就夠了，這里只所以用uint32類型的原因是它的操作必須是原子操作，只能使用原子操作支持的數據類型。

Do方法的實現方式
Do方法在一開始就會通過atomic.LoadUint32來獲取done字段的值，并且如果發現值為1就直接返回。這步只是初步保證了Do方法只會執行首次調用是傳入的函數。
不過單憑上面的判斷是不夠的。如果兩個goroutine都調用了同一個新的Once值的Do方法，并且幾乎同時執行到了其中的這個條件判斷代碼，那么它們就都會因判斷結果為false而繼續執行Do方法中剩余的代碼。
基于上面的可能，在初步保證的判斷之后，Do方法會立即鎖定其所屬值中的那個sync.Mutex類型的m字段。然后，它會在臨界區中再次檢查done字段的值。此時done的值應該仍然是0，并且已經加鎖。此時才認為是條件滿足，才會去調用參數函數。并且用原子操作把done的值變為1。

單例模式
如果熟悉設計模式中的單例模式的話，這個Do方法的實現方式，與單例模式有很多相似之處。都會先在臨界區之外判斷一次關鍵條件，若條件不滿足則立即返回。這通常被稱為快路徑，或者叫做快速失敗路徑。
如果條件滿足，那么到了臨界區中還要再對關鍵條件進行一次判斷，這主要是為了更加嚴謹。這兩次條件判斷常被統稱為（跨臨界區的）雙重檢查。由于進入臨界區前要加鎖，顯然會降低代碼的執行速度，所以其中的第二次條件判斷，以及后續的操作就被稱為慢路徑或者常規路徑。
Do方法中的代碼不多，但它卻應用了一個很經典的編程范式。

功能方面的特點

一、由于Do方法只會在參數函數執行結束之后把done字段的值變為1，因此，如果參數函數的執行需要很長的時間或者根本就不會結束，那么就有可能會導致相關goroutine的同時阻塞。
比如，有多個goroutine并發的調用了同一個Once值的Do方法，并且傳入的函數都會一直執行而不結束。那么，這些goroutine就都會因調用了這個Do方法而阻塞。此時，那個搶先執行了參數函數的goroutine之外，其他的goroutine都會被阻塞在該Once值的互斥鎖m的那行代碼上。
效果演示的示例代碼：

package main

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

func main() {
    once := sync.Once{}  // 這里換短變量聲明
    wg := sync.WaitGroup{}

    wg.Add(1)
    go func() {
        defer wg.Done()
        // 這個函數會被執行
        once.Do(func() {
            for i := 0; i < 10; i++ {
                fmt.Printf("\r任務[1-%d]執行中...", i)
                time.Sleep(time.Millisecond * 400)
            }
        })
        fmt.Printf("\n任務[1]執行完畢\n")
    }()

    wg.Add(1)
    go func() {
        defer wg.Done() 
        time.Sleep(time.Millisecond * 300)
        // 這句Do方法的調用會一直阻塞，知道上面的函數執行完畢
        // 然后Do方法里的函數不會執行
        once.Do(func() {
            fmt.Println("任務[2]執行中...")
        })
        // 上面Do方法阻塞結束后，直接會執行下面的代碼
        fmt.Println("任務[2]執行完畢")
    }()

    wg.Add(1)
    go func() {
        defer wg.Done() 
        time.Sleep(time.Millisecond * 300)
        once.Do(func() {
            fmt.Println("任務[3]執行中...")
        })
        fmt.Println("任務[3]執行完畢")
    }()

    wg.Wait()
    fmt.Println("Over")
}

二、Do方法在參數函數執行結束后，對done字段的賦值用的是原子操作，并且這一操作是被掛載defer語句中的。因此，不論參數函數的執行會以怎樣的方式結束，done字段的值都會變為1。
這樣就是說即時參數函數沒有執行成功，比如引發了panic。也是無法使用同一個Once值重新執行別的函數了。所以，如果需要為參數函數的執行設定重試機制，就要考慮在適當的時候替換Once值。
參考下面的示例：

package main

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

func main() {
    once := sync.Once{}
    wg := sync.WaitGroup{}

    wg.Add(1)
    go func() {
        defer wg.Done()
        defer func() {
            if p := recover(); p != nil {
                fmt.Printf("PANIC: %v\n", p)
                // 下面的語句會給once變量替換一個新的Once值，這樣下面的第二個任務還能被執行
                // once = sync.Once{}
            }
        }()
        once.Do(func() {
            fmt.Println("開始執行參數函數，緊接著會引發panic")
            panic(fmt.Errorf("主動引發了一個panic"))  // panic之后就去調用defer了
            fmt.Println("參數函數執行完畢")  // 這行不會執行，后面的都不會執行
        })
        fmt.Println("Do方法調用完畢")  // 這行也不會執行
    }()

    wg.Add(1)
    go func() {
        defer wg.Done() 
        time.Sleep(time.Millisecond * 500)
        once.Do(func() {
            fmt.Println("第二個任務執行中...")
            time.Sleep(time.Millisecond * 800)
            fmt.Println("第二個任務執行結束")
        })
        fmt.Println("第二個任務結束")
    }()

    wg.Wait()
    fmt.Println("Over")
}

延遲初始化

延遲一個昂貴的初始化步驟到有實際需求的時刻是一個很好的實踐。這也是sync.Once的一個使用場景。
下面是從書上改的示例代碼：

package main

import (
    "fmt"
    "sync"
)

var once sync.Once
var testmap map[string] int32

// 對testmap進行初始化的函數
func loadTestmap() {
    testmap = map[string] int32{
        "k1": 1,
        "k2": 2,
        "k3": 3,
    }
}

// 獲取testmap對應key的值，如果沒有初始化，會先執行初始化
// 書上說這個函數是并發安全的，這里的map初始化之后，內容不會再變
func getKey(key string) int32 {
    once.Do(loadTestmap)
    // 最后的return這句可能不是并發安全的，不過線程安全的map不是這里的重點
    // 假定這里的map在初始化之后只會被多個goroutine讀取，其內容不會再改變
    return testmap[key]
}

func main() {
    fmt.Println(getKey("k1"))
}

這里不考慮map線程安全的問題，而且書上的例子這里的map只用來存放數據，初始化之后不會對其內容進行修改。
這里主要是保證在變量初始化過程中的并發安全。以這種方式來使用sync.Once，可以避免變量在正確構造之前就被其它goroutine分享。否則，在別的goroutine中可能會獲取到一個內容不完整的變量。

總結

sync代碼包的WaitGroup類型和Once類型都是非常易用的同步工具。它們都是開箱即用和并發安全的。
Once類型使用互斥鎖和原子操作實現了功能，而WatiGroup類型中只用到了原子操作。所以可以說，它們都是更高層次的同步工具。它們都基于基本的同步工具，實現了某種特定的功能。sync包中的其他高級同步工具，其實也都是這樣的。