Go:有了 sync 為什麼還有 atomic?

2022-01-09 Go語言中文網

atomic

Go 是一種擅長並發的語言，啟動新的 goroutine 就像輸入 「go」 一樣簡單。隨著你發現自己構建的系統越來越複雜，正確保護對共享資源的訪問以防止競爭條件變得極其重要。此類資源可能包括可即時更新的配置（例如功能標誌）、內部狀態（例如斷路器狀態）等。

01 什麼是競態條件？

對於大多數讀者來說，這可能是基礎知識，但由於本文的其餘部分取決於對競態條件的理解，因此有必要進行簡短的複習。競態條件是一種情況，在這種情況下，程序的行為取決於其他不可控事件的順序或時間。在大多數情況下，這種情況是一個錯誤，因為可能會發生不希望的結果。

舉個具體的例子或許更容易理解：

// race_condition_test.go
package main

import (
 "fmt"
 "sort"
 "sync"
 "testing"
)

func Test_RaceCondition(t *testing.T) {
 var s = make([]int, 0)

 wg := sync.WaitGroup{}

 // spawn 10 goroutines to modify the slice in parallel
 for i := 0; i < 10; i++ {
  wg.Add(1)
  go func(i int) {
   defer wg.Done()
   s = append(s, i) //add a new item to the slice
  }(i)
 }

 wg.Wait()
 
 sort.Ints(s) //sort the response to have comparable results
 fmt.Println(s)
}
執行一：
$ go test -v race_condition_test.go
=== RUN   Test_RaceCondition
[0 1 2 3 4 5 6 7 8 9]
--- PASS: Test_RaceCondition (0.00s)
這裡看起來一切都很好。這是我們預期的輸出。該程序迭代了 10 次，並在每次迭代時將索引添加到切片中。
執行二：
=== RUN   Test_RaceCondition
[0 3]
--- PASS: Test_RaceCondition (0.00s)
等等，這裡發生了什麼？這次我們的響應切片中只有兩個元素。這是因為切片的內容 s 在加載和修改之間發生了變化，導致程序覆蓋了一些結果。這種特殊的競態條件是由數據競爭引起的，在這種情況下，多個 goroutine 嘗試同時訪問特定的共享變量，並且這些 goroutine 中的至少一個嘗試修改它。（注意，以上結果並非一定如此，每次運行結果可能都不相同）
如果你使用 -race 標誌執行測試，go 甚至會告訴你存在數據競爭並幫助你準確定位：
$ go test race_condition_test.go -race

==================
WARNING: DATA RACE
Read at 0x00c000132048 by goroutine 9:
  command-line-arguments.Test_RaceCondition.func1()
      /home/sfinlay/go/src/benchmarks/race_condition_test.go:20 +0xb4
  command-line-arguments.Test_RaceCondition·dwrap·1()
      /home/sfinlay/go/src/benchmarks/race_condition_test.go:21 +0x47

Previous write at 0x00c000132048 by goroutine 8:
  command-line-arguments.Test_RaceCondition.func1()
      /home/sfinlay/go/src/benchmarks/race_condition_test.go:20 +0x136
  command-line-arguments.Test_RaceCondition·dwrap·1()
      /home/sfinlay/go/src/benchmarks/race_condition_test.go:21 +0x47

Goroutine 9 (running) created at:
  command-line-arguments.Test_RaceCondition()
      /home/sfinlay/go/src/benchmarks/race_condition_test.go:18 +0xc5
  testing.tRunner()
      /usr/local/go/src/testing/testing.go:1259 +0x22f
  testing.(*T).Run·dwrap·21()
      /usr/local/go/src/testing/testing.go:1306 +0x47

Goroutine 8 (finished) created at:
  command-line-arguments.Test_RaceCondition()
      /home/sfinlay/go/src/benchmarks/race_condition_test.go:18 +0xc5
  testing.tRunner()
      /usr/local/go/src/testing/testing.go:1259 +0x22f
  testing.(*T).Run·dwrap·21()
      /usr/local/go/src/testing/testing.go:1306 +0x47
==================
02 並發控制
保護對這些共享資源的訪問通常涉及常見的內存同步機制，例如通道或互斥鎖。
這是將競態條件調整為使用互斥鎖的相同測試用例：
func Test_NoRaceCondition(t *testing.T) {
 var s = make([]int, 0)

 m := sync.Mutex{}
 wg := sync.WaitGroup{}

 // spawn 10 goroutines to modify the slice in parallel
 for i := 0; i < 10; i++ {
  wg.Add(1)
  go func(i int) {
   m.Lock()
   defer wg.Done()
   defer m.Unlock()
   s = append(s, i)
  }(i)
 }

 wg.Wait()

 sort.Ints(s) //sort the response to have comparable results
 fmt.Println(s)
}
這次它始終返回所有 10 個整數，因為它確保每個 goroutine 僅在沒有其他人執行時才讀寫切片。如果第二個 goroutine 同時嘗試獲取鎖，它必須等到前一個 goroutine 完成（即直到它解鎖）。
然而，對於高吞吐量系統，性能變得非常重要，因此減少鎖爭用（即一個進程或線程試圖獲取另一個進程或線程持有的鎖的情況）變得更加重要。執行此操作的最基本方法之一是使用讀寫鎖 ( sync.RWMutex) 而不是標準 sync.Mutex，但是 Go 還提供了一些原子內存原語即 atomic 包。
03 原子
Go 的 atomic 包提供了用於實現同步算法的低級原子內存原語。這聽起來像是我們需要的東西，所以讓我們嘗試用 atomic 重寫該測試：
import "sync/atomic"

func Test_RaceCondition_Atomic(t *testing.T) {
 var s = atomic.Value{}
 s.Store([]int{}) // store empty slice as the base

 wg := sync.WaitGroup{}

 // spawn 10 goroutines to modify the slice in parallel
 for i := 0; i < 10; i++ {
  wg.Add(1)
  go func(i int) {
   defer wg.Done()
   s1 := s.Load().([]int)
   s.Store(append(s1, i)) //replace the slice with a new one containing the new item
  }(i)
 }

 wg.Wait()

 s1 := s.Load().([]int)
 sort.Ints(s1) //sort the response to have comparable results
 fmt.Println(s1)
}
執行結果：
=== RUN   Test_RaceCondition_Atomic
[1 3]
--- PASS: Test_RaceCondition_Atomic (0.00s)
什麼？這和我們之前遇到的問題完全一樣，那麼這個包有什麼好處呢？
04 讀取-複製-更新
atomic 不是靈丹妙藥，它顯然不能替代互斥鎖，但是當涉及到可以使用讀取-複製-更新[1]模式管理的共享資源時，它非常出色。在這種技術中，我們通過引用獲取當前值，當我們想要更新它時，我們不修改原始值，而是替換指針（因此沒有人訪問另一個線程可能訪問的相同資源）。前面的示例無法使用此模式實現，因為它應該隨著時間的推移擴展現有資源而不是完全替換其內容，但在許多情況下，讀取-複製-更新是完美的。
這是一個基本示例，我們可以在其中獲取和存儲布爾值（例如，對於功能標誌很有用）。在這個例子中，我們正在執行一個並行基準測試，比較原子和讀寫互斥：
package main

import (
 "sync"
 "sync/atomic"
 "testing"
)

type AtomicValue struct{
 value atomic.Value
}

func (b *AtomicValue) Get() bool {
 return b.value.Load().(bool)
}

func (b *AtomicValue) Set(value bool) {
 b.value.Store(value)
}

func BenchmarkAtomicValue_Get(b *testing.B) {
 atomB := AtomicValue{}
 atomB.value.Store(false)

 b.RunParallel(func(pb *testing.PB) {
  for pb.Next() {
   atomB.Get()
  }
 })
}

/************/

type MutexBool struct {
 mutex sync.RWMutex
 flag  bool
}

func (mb *MutexBool) Get() bool {
 mb.mutex.RLock()
 defer mb.mutex.RUnlock()
 return mb.flag
}

func BenchmarkMutexBool_Get(b *testing.B) {
 mb := MutexBool{flag: true}

 b.RunParallel(func(pb *testing.PB) {
  for pb.Next() {
   mb.Get()
  }
 })
}
結果：
cpu: Intel(R) Core(TM) i7-8650U CPU @ 1.90GHz
BenchmarkAtomicValue_Get
BenchmarkAtomicValue_Get-8    1000000000          0.5472 ns/op
BenchmarkMutexBool_Get
BenchmarkMutexBool_Get-8      24966127            48.80 ns/op
結果很清楚。atomic 的速度提高了 89 倍以上。並且可以通過使用更原始的類型來進一步改進：
type AtomicBool struct{ flag int32 }

func (b *AtomicBool) Get() bool {
 return atomic.LoadInt32(&(b.flag)) != 0
}

func (b *AtomicBool) Set(value bool) {
 var i int32 = 0
 if value {
  i = 1
 }
 atomic.StoreInt32(&(b.flag), int32(i))
}

func BenchmarkAtomicBool_Get(b *testing.B) {
 atomB := AtomicBool{flag: 1}

 b.RunParallel(func(pb *testing.PB) {
  for pb.Next() {
   atomB.Get()
  }
 })
}
cpu: Intel(R) Core(TM) i7-8650U CPU @ 1.90GHz
BenchmarkAtomicBool_Get
BenchmarkAtomicBool_Get-8     1000000000          0.3161 ns/op
此版本比互斥鎖版本快 154 倍以上。
寫操作也顯示出明顯的差異（儘管規模並不那麼令人印象深刻）：
func BenchmarkAtomicBool_Set(b *testing.B) {
 atomB := AtomicBool{flag: 1}

 b.RunParallel(func(pb *testing.PB) {
  for pb.Next() {
   atomB.Set(true)
  }
 })
}

/************/

func BenchmarkAtomicValue_Set(b *testing.B) {
 atomB := AtomicValue{}
 atomB.value.Store(false)

 b.RunParallel(func(pb *testing.PB) {
  for pb.Next() {
   atomB.Set(true)
  }
 })
}

/************/

func BenchmarkMutexBool_Set(b *testing.B) {
 mb := MutexBool{flag: true}

 b.RunParallel(func(pb *testing.PB) {
  for pb.Next() {
   mb.Set(true)
  }
 })
}
結果：
cpu: Intel(R) Core(TM) i7-8650U CPU @ 1.90GHz
BenchmarkAtomicBool_Set
BenchmarkAtomicBool_Set-8     64624705         16.79 ns/op
BenchmarkAtomicValue_Set
BenchmarkAtomicValue_Set-8    47654121         26.43 ns/op
BenchmarkMutexBool_Set
BenchmarkMutexBool_Set-8      20124637         66.50 ns/op
在這裡我們可以看到 atomic 在寫入時比在讀取時慢得多，但仍然比互斥鎖快得多。有趣的是，我們可以看到互斥鎖讀取和寫入之間的差異不是很明顯（慢 30%）。儘管如此， atomic 仍然表現得更好（比互斥鎖快 2-4 倍）。
05 為什麼 atomic 這麼快？
簡而言之，原子操作很快，因為它們依賴於原子 CPU 指令而不是依賴外部鎖。使用互斥鎖時，每次獲得鎖時，goroutine 都會短暫暫停或中斷，這種阻塞佔使用互斥鎖所花費時間的很大一部分。原子操作可以在沒有任何中斷的情況下執行。
06 atomic 總是答案嗎？
正如我們在一個早期示例中已經證明的那樣，atomic 無法解決所有問題，某些操作只能使用互斥鎖來解決。
考慮以下示例，該示例演示了我們使用 map 作為內存緩存的常見模式：
package main

import (
 "sync"
 "sync/atomic"
 "testing"
)

//Don't use this implementation!
type AtomicCacheMap struct {
 value atomic.Value //map[int]int
}

func (b *AtomicCacheMap) Get(key int) int {
 return b.value.Load().(map[int]int)[key]
}

func (b *AtomicCacheMap) Set(key, value int) {
 oldMap := b.value.Load().(map[int]int)
 newMap := make(map[int]int, len(oldMap)+1)
 for k, v := range oldMap {
  newMap[k] = v
 }
 newMap[key] = value
 b.value.Store(newMap)
}

func BenchmarkAtomicCacheMap_Get(b *testing.B) {
 atomM := AtomicCacheMap{}
 atomM.value.Store(testMap)

 b.RunParallel(func(pb *testing.PB) {
  for pb.Next() {
   atomM.Get(0)
  }
 })
}

func BenchmarkAtomicCacheMap_Set(b *testing.B) {
 atomM := AtomicCacheMap{}
 atomM.value.Store(testMap)

 var i = 0
 b.RunParallel(func(pb *testing.PB) {
  for pb.Next() {
   atomM.Set(i, i)
   i++
  }
 })
}

/************/

type MutexCacheMap struct {
 mutex sync.RWMutex
 value map[int]int
}

func (mm *MutexCacheMap) Get(key int) int {
 mm.mutex.RLock()
 defer mm.mutex.RUnlock()
 return mm.value[key]
}

func (mm *MutexCacheMap) Set(key, value int) {
 mm.mutex.Lock()
 defer mm.mutex.Unlock()
 mm.value[key] = value
}

func BenchmarkMutexCacheMap_Get(b *testing.B) {
 mb := MutexCacheMap{value: testMap}

 b.RunParallel(func(pb *testing.PB) {
  for pb.Next() {
   mb.Get(0)
  }
 })
}

func BenchmarkMutexCacheMap_Set(b *testing.B) {
 mb := MutexCacheMap{value: testMap}

 var i = 0
 b.RunParallel(func(pb *testing.PB) {
  for pb.Next() {
   mb.Set(i, i)
   i++
  }
 })
}
結果：
cpu: Intel(R) Core(TM) i7-8650U CPU @ 1.90GHz
BenchmarkAtomicCacheMap_Get
BenchmarkAtomicCacheMap_Get-8    301664540           4.194 ns/op
BenchmarkAtomicCacheMap_Set
BenchmarkAtomicCacheMap_Set-8       87637            95889 ns/op
BenchmarkMutexCacheMap_Get
BenchmarkMutexCacheMap_Get-8     20000959            54.63 ns/op
BenchmarkMutexCacheMap_Set
BenchmarkMutexCacheMap_Set-8      5012434            267.2 ns/op
哎呀，這種表現是痛苦的。這意味著，當必須複製大型結構時，atomic 的性能非常差。不僅如此，此代碼還包含競態條件。就像本文開頭的切片案例一樣，原子緩存示例具有競態條件，其中可能會在複製 map 和存儲 map 的時間之間添加新的緩存條目，在這種情況下，新條目將丟失。在這種情況下，該 -race 標誌不會檢測到任何數據競爭，因為沒有對同一 map 的並發訪問。
07 注意事項
Go 的文檔[2]警告了 atomic 包的潛在誤用：
這些函數需要非常小心才能正確使用。除了特殊的低級應用程式，同步最好使用通道或 sync 包的工具來完成。通過通信共享內存；不要通過共享內存進行通信。
開始使用 atomic 包時，你可能會遇到的第一個問題是：
panic: sync/atomic: store of inconsistently typed value into Value
使用 atomic.Store，確保每次調用方法時都存儲完全相同的類型很重要。這聽起來很容易，但通常並不像聽起來那麼簡單：
package main

import (
 "fmt"
 "sync/atomic"
)

//Our own custom error type which implements the error interface
type CustomError struct {
 Code    int
 Message string
}

func (e CustomError) Error() string {
 return fmt.Sprintf("%d: %s", e.Code, e.Message)
}

func InternalServerError(msg string) error {
 return CustomError{Code: 500, Message: msg}
}

func main() {
 var (
  err1 error = fmt.Errorf("error happened")
  err2 error = InternalServerError("another error happened")
 )

 errVal := atomic.Value{}
 errVal.Store(err1)
 errVal.Store(err2) //panics here
}
兩個值都是 error 類型是不夠的，因為它們只是實現了錯誤接口。它們的具體類型仍然不同，因此 atomic 不喜歡它。
08 總結
競態條件很糟糕，應該保護對共享資源的訪問。互斥體很酷，但由於鎖爭用而趨於緩慢。對於某些讀取-複製-更新模式有意義的情況（這往往是動態配置之類的東西，例如特性標誌、日誌級別或 map 或結構體，一次填充例如通過 JSON 解析等），尤其是當讀取次數比寫入次數多時。atomic 通常不應用於其他用例（例如，隨時間增長的變量，如緩存），並且該特性的使用需要非常小心。
可能最重要的方法是將鎖保持在最低限度，如果你在在考慮原子等替代方案，請務必在投入生產之前對其進行廣泛的測試和試驗。
原文連結：https://www.sixt.tech/golangs-atomic
參考資料[1]
讀取-複製-更新: https://en.wikipedia.org/wiki/Read-copy-update
[2]
文檔: https://pkg.go.dev/sync/atomic

相關焦點

  • Golang中sync.Map的實現原理
    需要並發讀寫時，一般的做法是加鎖，但這樣性能並不高，Go語言在 1.9 版本中提供了一種效率較高的並發安全的 sync.Map，今天，我們就來講講 sync.Map的用法以及原理#  使用方法```gofunc main() { var m sync.Map //插入 m.Store("1","a")
  • Go 並發編程 — sync.Pool 源碼級原理剖析
    ，有可能是新的（新分配的），有可能是舊的（之前人用過，然後 Put 進去的）；不能對 Pool 池裡的元素個數做假定，你不能夠；sync.Pool 本身的 Get, Put 調用是並發安全的，sync.New 指向的初始化函數會並發調用，裡面安不安全只有自己知道；當用完一個從 Pool 取出的實例時候，一定要記得調用 Put，否則 Pool 無法復用這個實例，通常這個用 defer 完成；
  • 詳解並發編程之sync.Once的實現(附上三道面試題)
    上一篇我們一起分析了atomic包，今天我們一起來看一看sync/once的使用與實現.什麼是sync.onceGo語言標準庫中的sync.Once可以保證go程序在運行期間的某段代碼只會執行一次，作用與init類似，但是也有所不同：init函數是在文件包首次被加載的時候執行，且只執行一次。
  • 源碼剖析sync.Cond(條件變量的實現機制)
    sync.Cond的基本使用Go標準庫提供了Cond原語，為等待/通知場景下的並發問題提供支持。先看這樣一個例子：var ( done = false topic = "Golang夢工廠")func main() { cond := sync.NewCond(&sync.Mutex{}) go Consumer
  • Go Mutex 之4種易錯場景盤點
    不過 Go 有死鎖檢查機制，如果出現了這種情況，直接運行，是會報錯的。除此以外，還可以用 go vet 工具主動檢查死鎖。defer m.Unlock() copyF(m) return}直接 go run 會報錯，通過 go vet main.go，結果如下：main.go:11:14: copyF passes lock by value: sync.Mutexmain.go:26:8:
  • 『每周譯Go』Go sync.Once 的妙用
    sync.WaitGroup 或是 sync.Map，但是你聽說過 sync.Once 麼？也許你聽說過，那 go 文檔是怎麼描述它的呢？Once 是只執行一個操作的對象。聽起來很簡單，它有什麼用處呢？由於某些原因，sync.Once 的用法並沒有很好的文檔記錄。
  • 為什麼 Go map 和 slice 是非線性安全的?
    其中最常見的初級錯誤，Go 面試較最愛問的問題之一：為什麼在 Go 語言裡，map 和 slice 不支持並發讀寫，也就是是非線性安全的，為什麼不支持？見招拆招後，緊接著就會開始討論如何讓他們倆 」冤家「 支持並發讀寫？
  • Go 並發編程之 Mutex
    【導讀】go 並發場景下如何保障數據讀寫正確？本文梳理了 Mutex 的用法。go mutex 臨界區示意圖Mutex 是 Go 語言中使用最廣泛的同步原語，也稱為並發原語，解決的是並發讀寫共享資源，避免出現數據競爭 data race 問題。
  • 看了這篇 Go map 並發不用慌
    分別是：有了選擇，總是有選擇困難症的，這兩種到底怎麼選，誰的性能更加的好？我有一個朋友說 標準庫 sync.Map 性能菜的很，不要用。我到底聽誰的...sync.Map 剖析清楚如何測試，測試的結果後。我們需要進一步深挖，知其所以然。為什麼 sync.Map 類型的測試結果這麼的 「偏科」，為什麼讀操作性能這麼高，寫操作性能低的可怕，他是怎麼設計的？
  • 線程安全之std::atomic探索
    多線程的數據訪問一直很讓程式設計師們頭大，有什麼簡便方法來實現多線程間的通呢？
  • Go 並發任務編排利器之 WaitGroup
    var m sync.Mutex var wg sync.WaitGroup var count int wg.Add(1000) for i := 0; i < 1000; i++ { go compute(&m, &wg, i*1000+1, (i+1)*1000+1, &count) } wg.Wait
  • go-redis 連接池實現思想
    為什麼會有redis連接池的概念    在傳統的C/S模型中，客戶端和服務端需要通信的話就要建立連接，在這裡每次連接都需要花費不少的資源，例如常說的TCP三次握手等。    連接池則可以實現在客戶端建立多個連結並且不釋放，當需要使用連結的時候通過一定的算法獲取已經建立的連結，使用完了以後則還給連接池，這就免去了TCP建立連接所帶來的系統調用所消耗的時間。上面都是瞎BB，不想看的直接跳到下面看正文。
  • Go 性能分析工具 pprof 入門
    支持的 CMD 有:goroutine: 獲取程序當前所有 goroutine 的堆棧信息。heap: 包含每個 goroutine 分配大小，分配堆棧等。每分配 runtime.MemProfileRate(默認為512K) 個字節進行一次數據採樣。
  • Golang語言標準庫 sync 包的 Once 怎麼使用?
    ，sync 包有一個 Once 類型，官方文檔介紹 Once 是一個只執行一次操作的對象。03實現原理sync.Once 源碼：type Once struct { done uint32 m Mutex}func (o *Once) Do(f func()) {  if atomic.LoadUint32(&o.done
  • 淺談 Golang 鎖的應用: sync包
    實際開發中，鎖又可分為互斥鎖（排它鎖）、讀寫鎖、共享鎖、自旋鎖，甚至還有悲觀鎖、樂觀鎖這種說法。在 Mysql 資料庫裡面鎖的應用更多，比如行鎖、表鎖、間隙鎖，有點眼花繚亂。這些鎖的機制在 Go 裡面有什麼應用呢，下面大家一起看看 Go 標準庫裡面 sync 包提供的一些非常強大的基於鎖的實現。1.
  • 我與sync.Once的愛恨糾纏
    sync.Once 的用法 在多數情況下，sync.Once 被用於控制變量的初始化，這個變量的讀寫通常遵循單例模式，滿足這三個條件：變量被初始化過程中，所有讀都被阻塞（讀操作；當變量初始化完成後，讀操作繼續進行）在標準庫中不乏有大量 sync.Once 的使用案例，在 strings
  • Go 日誌庫 zerolog 大解剖
    在這個時候，需要有一個對象池，程序不再自己單獨創建對象，而是從對象池中獲取。使用 sync.Pool 可以將暫時不用的對象緩存起來，下次需要的時候從池中取，不用再次經過內存分配。這裡有個連結，通過這個 issue 23199[4]了解到使用動態增長的 buffer 會導致大量內存被固定，在活鎖的情況下永遠不會釋放。