併發安全Sync包的使用

有時候在Go代碼中可能會存在多個goroutine同時操做一個資源（臨界區），這種狀況會發生競態問題（數據競態）。Sync包主要實現了併發任務同步WaitGroup的幾種方法和併發安全的互斥鎖和讀寫鎖方法，還實現了比較特殊的兩個方法，一個是保持只執行一次的Once方法和線程安全的Map。

sync.WaitGroup(同步等待)

sync.WaitGroup內部維護着一個計數器Add()，計數器的值能夠增長和減小。例如當咱們啓動了N 個併發任務時，就將計數器值增長N。每一個任務完成時經過調用Done()方法將計數器減1，底層爲Add(-1)。經過調用Wait()來等待併發任務執行完，當計數器值爲0時，表示全部併發任務已經完成。

var x int64
var wg sync.WaitGroup

func add() {
    for i := 0; i < 5000; i++ {
        x = x + 1 //數據競爭
    }
    wg.Done()
}
func main() {
    wg.Add(2)
    go add()
    go add()
    wg.Wait()
    fmt.Println(x)
}

上面的代碼中咱們開啓了兩個goroutine去累加變量x的值，這兩個goroutine在訪問和修改x變量的時候就會存在數據競爭，致使最後的結果與期待的不符。

sync.Mutex(互斥鎖)

互斥鎖是一種經常使用的控制共享資源訪問的方法，它可以保證同時只有一個goroutine能夠訪問共享資源。Go語言中使用sync包的Mutex類型來實現互斥鎖。 使用互斥鎖來修復上面代碼的問題：

var x int64
var wg sync.WaitGroup
var lock sync.Mutex

func add() {
    for i := 0; i < 5000; i++ {
        lock.Lock() // 加鎖
        x = x + 1
        lock.Unlock() // 解鎖
    }
    wg.Done()
}
func main() {
    wg.Add(2)
    go add()
    go add()
    wg.Wait()
    fmt.Println(x)
}

使用互斥鎖可以保證同一時間有且只有一個goroutine進入臨界區，其餘的goroutine則在等待鎖；當互斥鎖釋放後，等待的goroutine才能夠獲取鎖進入臨界區，多個goroutine同時等待一個鎖時，喚醒的策略是隨機的。

sync.RWMutex(讀寫互斥鎖)

互斥鎖是徹底互斥的，可是有不少實際的場景下是讀多寫少的，當咱們併發的去讀取一個資源不涉及資源修改的時候是沒有必要加鎖的，這種場景下使用讀寫鎖是更好的一種選擇。讀寫鎖在Go語言中使用sync包中的RWMutex類型。

讀寫鎖分爲兩種：讀鎖和寫鎖。當一個goroutine獲取讀鎖以後，其餘的goroutine若是是獲取讀鎖會繼續得到鎖，若是是獲取寫鎖就會等待；當一個goroutine獲取寫鎖以後，其餘的goroutine不管是獲取讀鎖仍是寫鎖都會等待。

讀寫鎖示例：

var (
    x      int64
    wg     sync.WaitGroup
    lock   sync.Mutex
    rwlock sync.RWMutex
)

func write() {
    // lock.Lock()   // 加互斥鎖
    rwlock.Lock() // 加寫鎖
    x = x + 1
    time.Sleep(10 * time.Millisecond) // 假設讀操做耗時10毫秒
    rwlock.Unlock()                   // 解寫鎖
    // lock.Unlock()                     // 解互斥鎖
    wg.Done()
}

func read() {
    // lock.Lock()                  // 加互斥鎖
    rwlock.RLock()               // 加讀鎖
    time.Sleep(time.Millisecond) // 假設讀操做耗時1毫秒
    rwlock.RUnlock()             // 解讀鎖
    // lock.Unlock()                // 解互斥鎖
    wg.Done()
}

func main() {
    start := time.Now()
    for i := 0; i < 10; i++ {
        wg.Add(1)
        go write()
    }

    for i := 0; i < 1000; i++ {
        wg.Add(1)
        go read()
    }

    wg.Wait()
    end := time.Now()
    fmt.Println(end.Sub(start))
}

須要注意的是讀寫鎖很是適合讀多寫少的場景，若是讀和寫的操做差異不大，讀寫鎖的優點就發揮不出來。

sync.Once(單例)

說在前面的話：這是一個進階知識點。

在編程的不少場景下咱們須要確保某些操做在高併發的場景下只執行一次，例如只加載一次配置文件、只關閉一次通道等。

Go語言中的sync包中提供了一個針對只執行一次場景的解決方案–sync.Once。

sync.Once只有一個Do方法，其簽名以下：

func (o *Once) Do(f func()) {}

注意：若是要執行的函數f須要傳遞參數就須要搭配閉包來使用。

sync.Map(線程安全map)

Go語言中內置的map不是併發安全的。請看下面的示例：

var m = make(map[string]int)

func get(key string) int {
    return m[key]
}

func set(key string, value int) {
    m[key] = value
}

func main() {
    wg := sync.WaitGroup{}
    for i := 0; i < 20; i++ {
        wg.Add(1)
        go func(n int) {
            key := strconv.Itoa(n)
            set(key, n)
            fmt.Printf("k=:%v,v:=%v\n", key, get(key))
            wg.Done()
        }(i)
    }
    wg.Wait()
}

上面的代碼開啓少許幾個goroutine的時候可能沒什麼問題，當併發多了以後執行上面的代碼就會報fatal error: concurrent map writes錯誤。

像這種場景下就須要爲map加鎖來保證併發的安全性了，Go語言的sync包中提供了一個開箱即用的併發安全版map–sync.Map。開箱即用表示不用像內置的map同樣使用make函數初始化就能直接使用。同時sync.Map內置了諸如Store、Load、LoadOrStore、Delete、Range等操做方法。

var m = sync.Map{}

func main() {
    wg := sync.WaitGroup{}
    for i := 0; i < 20; i++ {
        wg.Add(1)
        go func(n int) {
            key := strconv.Itoa(n)
            m.Store(key, n)
            value, _ := m.Load(key)
            fmt.Printf("k=:%v,v:=%v\n", key, value)
            wg.Done()
        }(i)
    }
    wg.Wait()
}

sync/atomic(原子操做)

代碼中的加鎖操做由於涉及內核態的上下文切換會比較耗時、代價比較高。針對基本數據類型咱們還可使用原子操做來保證併發安全，由於原子操做是Go語言提供的方法它在用戶態就能夠完成，所以性能比加鎖操做更好。Go語言中原子操做由內置的標準庫sync/atomic提供。

方法	操做類型
func LoadInt32(addr *int32) (val int32)	讀取操做
func LoadInt64(addr *int64) (val int64)	讀取操做
func LoadUint32(addr *uint32) (val uint32)	讀取操做
func LoadUint64(addr *uint64) (val uint64)	讀取操做
func LoadUintptr(addr *uintptr) (val uintptr)	讀取操做
func LoadPointer(addr *unsafe.Pointer) (val unsafe.Pointer)	讀取操做
func StoreInt32(addr *int32, val int32)	寫入操做
func StoreInt64(addr *int64, val int64)	寫入操做
func StoreUint32(addr *uint32, val uint32)	寫入操做
func StoreUint64(addr *uint64, val uint64)	寫入操做
func StoreUintptr(addr *uintptr, val uintptr)	寫入操做
func StorePointer(addr *unsafe.Pointer, val unsafe.Pointer)	寫入操做
func AddInt32(addr *int32, delta int32) (new int32)	修改操做
func AddInt64(addr *int64, delta int64) (new int64)	修改操做
func AddUint32(addr *uint32, delta uint32) (new uint32)	修改操做
func AddUint64(addr *uint64, delta uint64) (new uint64)	修改操做
func AddUintptr(addr *uintptr, delta uintptr) (new uintptr)	修改操做
func SwapInt32(addr *int32, new int32) (old int32)	交換操做
func SwapInt64(addr *int64, new int64) (old int64)	交換操做
func SwapUint32(addr *uint32, new uint32) (old uint32)	交換操做
func SwapUint64(addr *uint64, new uint64) (old uint64)	交換操做
func SwapUintptr(addr *uintptr, new uintptr) (old uintptr)	交換操做
func SwapPointer(addr *unsafe.Pointer, new unsafe.Pointer) (old unsafe.Pointer)	交換操做
func CompareAndSwapInt32(addr *int32, old, new int32) (swapped bool)	比較並交換操做
func CompareAndSwapInt64(addr *int64, old, new int64) (swapped bool)	比較並交換操做
func CompareAndSwapUint32(addr *uint32, old, new uint32) (swapped bool)	比較並交換操做
func CompareAndSwapUint64(addr *uint64, old, new uint64) (swapped bool)	比較並交換操做
func CompareAndSwapUintptr(addr *uintptr, old, new uintptr) (swapped bool)	比較並交換操做
func CompareAndSwapPointer(addr *unsafe.Pointer, old, new unsafe.Pointer) (swapped bool)	比較並交換操做

示例：

package main

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

var x int64
var l sync.Mutex
var wg sync.WaitGroup

// 普通版加函數
func add() {
    // x = x + 1
    x++ // 等價於上面的操做
    wg.Done()
}

// 互斥鎖版加函數
func mutexAdd() {
    l.Lock()
    x++
    l.Unlock()
    wg.Done()
}

// 原子操做版加函數
func atomicAdd() {
    atomic.AddInt64(&x, 1)
    wg.Done()
}

func main() {
    start := time.Now()
    for i := 0; i < 10000; i++ {
        wg.Add(1)
        //go add()       // 普通版add函數 不是併發安全的
        //go mutexAdd()  // 加鎖版add函數 是併發安全的，可是加鎖性能開銷大
        go atomicAdd() // 原子操做版add函數 是併發安全，性能優於加鎖版
    }
    wg.Wait()
    end := time.Now()
    fmt.Println(x)
    fmt.Println(end.Sub(start))
}