寫在前面

本文基于 Golang 1.14

Go 提供了 channel 或 mutex 等內(nèi)存同步機(jī)制，有助于解決不同的問題。在共享內(nèi)存的情況下，mutex 可以保護(hù)內(nèi)存不發(fā)生數(shù)據(jù)競(jìng)爭(zhēng)(data race)。不過，雖然存在兩個(gè) mutex，但 Go 也通過 atomic 包提供了原子內(nèi)存基元來提高性能。在深入研究解決方案之前，我們先回過頭來看看數(shù)據(jù)競(jìng)爭(zhēng)。

數(shù)據(jù)競(jìng)爭(zhēng)

當(dāng)兩個(gè)或兩個(gè)以上的 goroutine 同時(shí)訪問同一塊內(nèi)存區(qū)域，并且其中至少有一個(gè)在寫時(shí)，就會(huì)發(fā)生數(shù)據(jù)競(jìng)爭(zhēng)。雖然 map 內(nèi)部有一定的機(jī)制來防止數(shù)據(jù)競(jìng)爭(zhēng)，但一個(gè)簡(jiǎn)單的結(jié)構(gòu)體并沒有任何的機(jī)制，因此容易發(fā)生數(shù)據(jù)競(jìng)爭(zhēng)。

為了說明數(shù)據(jù)競(jìng)爭(zhēng)，我以一個(gè)goroutine 持續(xù)更新的配置為例向大家展示一下。 

package main 
 
import ( 
    "fmt" 
    "sync" 
) 
 
type Config struct { 
    a []int 
} 
 
func main() { 
    cfg := &Config{} 
 
    // 啟動(dòng)一個(gè) writer goroutine，不斷寫入數(shù)據(jù) 
    go func() { 
        i := 0 
 
        for { 
            i++ 
            cfg.a = []int{i, i + 1, i + 2, i + 3, i + 4, i + 5} 
        } 
    }() 
 
    // 啟動(dòng)多個(gè) reader goroutine，不斷獲取數(shù)據(jù) 
    var wg sync.WaitGroup 
    for n := 0; n < 4; n++ { 
        wg.Add(1) 
        go func() { 
            for n := 0; n < 100; n++ { 
                fmt.Printf("%#v\n", cfg) 
            } 
            wg.Done() 
        }() 
    } 
 
    wg.Wait() 
} 

運(yùn)行這段代碼可以清楚地看到，原本期望是運(yùn)行上述代碼后，每一行的數(shù)字應(yīng)該是連續(xù)的，但是由于數(shù)據(jù)競(jìng)爭(zhēng)的存在，導(dǎo)致結(jié)果是非確定性的。 

F:\hello>go run main.go 
[...] 
&main.Config{a:[]int{180954, 180962, 180967, 180972, 180977, 180983}} 
&main.Config{a:[]int{181296, 181304, 181311, 181318, 181322, 181323}} 
&main.Config{a:[]int{181607, 181617, 181624, 181631, 181636, 181643}} 

我們可以在運(yùn)行時(shí)加入?yún)?shù) --race 看一下結(jié)果： 

F:\hello>go run --race main.go 
[...] 
&main.Config{a:[]int(nil)} 
================== 
&main.Config{a:[]int(nil)} 
WARNING: DATA RACE&main.Config{a:[]int(nil)} 
 
Read at 0x00c00000c210 by Goroutine 9: 
  reflect.Value.Int() 
      D:/Go/src/reflect/value.go:988 +0x3584 
  fmt.(*pp).printValue() 
      D:/Go/src/fmt/print.go:749 +0x3590 
  fmt.(*pp).printValue() 
      D:/Go/src/fmt/print.go:860 +0x8f2 
  fmt.(*pp).printValue() 
      D:/Go/src/fmt/print.go:810 +0x289a 
  fmt.(*pp).printValue() 
      D:/Go/src/fmt/print.go:880 +0x261c 
  fmt.(*pp).printArg() 
      D:/Go/src/fmt/print.go:716 +0x26b 
  fmt.(*pp).doPrintf() 
      D:/Go/src/fmt/print.go:1030 +0x326 
  fmt.Fprintf() 
      D:/Go/src/fmt/print.go:204 +0x86 
  fmt.Printf() 
      D:/Go/src/fmt/print.go:213 +0xbc 
  main.main.func2() 
      F:/hello/main.go:31 +0x42 
 
Previous write at 0x00c00000c210 by goroutine 7: 
  main.main.func1() 
      F:/hello/main.go:21 +0x66 
 
goroutine 9 (running) created at: 
  main.main() 
      F:/hello/main.go:29 +0x124 
 
goroutine 7 (running) created at: 
  main.main() 
      F:/hello/main.go:16 +0x95 
================== 

為了避免同時(shí)讀寫過程中產(chǎn)生的數(shù)據(jù)競(jìng)爭(zhēng)最常采用的方法可能是使用 mutex 或 atomic 包。

Mutex?還是 Atomic?

標(biāo)準(zhǔn)庫在 sync 包提供了兩種互斥鎖 ：sync.Mutex 和 sync.RWMutex。后者在你的程序需要處理多個(gè)讀操作和極少的寫操作時(shí)進(jìn)行了優(yōu)化。

針對(duì)上面代碼中產(chǎn)生的數(shù)據(jù)競(jìng)爭(zhēng)問題，我們看一下，如何解決呢?

使用 sync.Mutex 解決數(shù)據(jù)競(jìng)爭(zhēng) 

package main 
 
import ( 
    "fmt" 
    "sync" 
) 
 
// Config 定義一個(gè)結(jié)構(gòu)體用于模擬存放配置數(shù)據(jù) 
type Config struct { 
    a []int 
} 
 
func main() { 
    cfg := &Config{} 
    var mux sync.RWMutex 
 
    // 啟動(dòng)一個(gè) writer goroutine，不斷寫入數(shù)據(jù) 
    go func() { 
        i := 0 
 
        for { 
            i++ 
            // 進(jìn)行數(shù)據(jù)寫入時(shí)，先通過鎖進(jìn)行鎖定 
            mux.Lock() 
            cfg.a = []int{i, i + 1, i + 2, i + 3, i + 4, i + 5} 
            mux.Unlock() 
        } 
    }() 
 
    // 啟動(dòng)多個(gè) reader goroutine，不斷獲取數(shù)據(jù) 
    var wg sync.WaitGroup 
    for n := 0; n < 4; n++ { 
        wg.Add(1) 
        go func() { 
            for n := 0; n < 100; n++ { 
                // 因?yàn)檫@里只是需要讀取數(shù)據(jù)，所以只需要加一個(gè)讀鎖即可 
                mux.RLock() 
                fmt.Printf("%#v\n", cfg) 
                mux.RUnlock() 
            } 
            wg.Done() 
        }() 
    } 
 
    wg.Wait() 
} 

通過上面的代碼，我們做了兩處改動(dòng)。第一處改動(dòng)在寫數(shù)據(jù)前通過 mux.Lock() 加了一把鎖;第二處改動(dòng)在讀數(shù)據(jù)前通過 mux.RLock() 加了一把讀鎖。

運(yùn)行上述代碼看一下結(jié)果： 

F:\hello>go run --race main.go 
&main.Config{a:[]int{512, 513, 514, 515, 516, 517}} 
&main.Config{a:[]int{512, 513, 514, 515, 516, 517}} 
&main.Config{a:[]int{513, 514, 515, 516, 517, 518}} 
&main.Config{a:[]int{513, 514, 515, 516, 517, 518}} 
&main.Config{a:[]int{513, 514, 515, 516, 517, 518}} 
&main.Config{a:[]int{513, 514, 515, 516, 517, 518}} 
&main.Config{a:[]int{514, 515, 516, 517, 518, 519}} 
[...] 

這次達(dá)到了我們的預(yù)期并且也沒有產(chǎn)生數(shù)據(jù)競(jìng)爭(zhēng)。

使用 atomic 解決數(shù)據(jù)競(jìng)爭(zhēng) 

package main 
 
import ( 
    "fmt" 
    "sync" 
    "sync/atomic" 
) 
 
type Config struct { 
    a []int 
} 
 
func main() { 
    var v atomic.Value 
 
    // 寫入數(shù)據(jù) 
    go func() { 
        var i int 
        for { 
            i++ 
            cfg := Config{ 
                a: []int{i, i + 1, i + 2, i + 3, i + 4, i + 5}, 
            } 
            v.Store(cfg) 
        } 
    }() 
 
    // 讀取數(shù)據(jù) 
    var wg sync.WaitGroup 
    for n := 0; n < 4; n++ { 
        wg.Add(1) 
        go func() { 
            for n := 0; n < 100; n++ { 
                cfg := v.Load() 
                fmt.Printf("%#v\n", cfg) 
            } 
            wg.Done() 
        }() 
    } 
 
    wg.Wait() 
} 

這里我們使用了 atomic 包，通過運(yùn)行我們發(fā)現(xiàn)，也同樣達(dá)到了我們期望的結(jié)果： 

[...] 
main.Config{a:[]int{219142, 219143, 219144, 219145, 219146, 219147}} 
main.Config{a:[]int{219491, 219492, 219493, 219494, 219495, 219496}} 
main.Config{a:[]int{219826, 219827, 219828, 219829, 219830, 219831}} 
main.Config{a:[]int{219948, 219949, 219950, 219951, 219952, 219953}} 

從生成的輸出結(jié)果而言，看起來使用 atomic 包的解決方案要快得多，因?yàn)樗梢陨筛叩臄?shù)字序列。為了更加嚴(yán)謹(jǐn)?shù)淖C明這個(gè)結(jié)果，我們下面將對(duì)這兩個(gè)程序進(jìn)行基準(zhǔn)測(cè)試。

性能分析

一個(gè) benchmark 應(yīng)該根據(jù)被測(cè)量的內(nèi)容來解釋。因此，我們假設(shè)之前的程序，有一個(gè)不斷存儲(chǔ)新配置的 數(shù)據(jù)寫入器，同時(shí)也有多個(gè)不斷讀取配置的 數(shù)據(jù)讀取器。為了涵蓋更多潛在的場(chǎng)景，我們還將包括一個(gè)只有 數(shù)據(jù)讀取器 的 benchmark，假設(shè) Config 不經(jīng)常改變。

下面是部分 benchmark 的代碼： 

func BenchmarkMutexMultipleReaders(b *testing.B) { 
    var lastValue uint64 
    var mux sync.RWMutex 
    var wg sync.WaitGroup 
 
    cfg := Config{ 
        a: []int{0, 0, 0, 0, 0, 0}, 
    } 
 
    for n := 0; n < 4; n++ { 
        wg.Add(1) 
 
        go func() { 
            for n := 0; n < b.N; n++ { 
                mux.RLock() 
                atomic.SwapUint64(&lastValue, uint64(cfg.a[0])) 
                mux.RUnlock() 
            } 
            wg.Done() 
        }() 
    } 
 
    wg.Wait() 
} 

執(zhí)行上面的測(cè)試代碼后我們可以得到如下的結(jié)果： 

name                              time/op 
AtomicOneWriterMultipleReaders-4  72.2ns ± 2% 
AtomicMultipleReaders-4           65.8ns ± 2% 
 
MutexOneWriterMultipleReaders-4    717ns ± 3% 
MutexMultipleReaders-4             176ns ± 2% 

基準(zhǔn)測(cè)試證實(shí)了我們之前看到的性能情況。為了了解 mutex 的瓶頸到底在哪里，我們可以在啟用 tracer 的情況下重新運(yùn)行程序。

goroutines 運(yùn)行時(shí)不間斷，能夠完成任務(wù)。對(duì)于帶有 mutex 的程序的配置文件，得到的結(jié)果那是完全不同的。

現(xiàn)在運(yùn)行時(shí)間相當(dāng)零碎，這是由于停放 goroutine 的 mutex 造成的。這一點(diǎn)可以從 goroutine 的概覽中得到證實(shí)，其中顯示了同步時(shí)被阻塞的時(shí)間。

屏蔽時(shí)間大概占到三分之一的時(shí)間，這一點(diǎn)可以從下面的 block profile 的圖中詳細(xì)看到。 

在這種情況下，atomic 包肯定會(huì)帶來優(yōu)勢(shì)。但是，在某些方面可能會(huì)降低性能。例如，如果你要存儲(chǔ)一張大地圖，每次更新地圖時(shí)都要復(fù)制它，這樣效率就很低。

via: https://medium.com/a-journey-with-go/go-how-to-reduce-lock-contention-with-the-atomic-package-ba3b2664b549

作者：Vincent Blanchon 譯者：double12gzh 校對(duì)：lxbwolf