Go并發(fā)編程結(jié)構(gòu)體多字段原子操作示例詳解

更新時(shí)間：2023年12月01日 10:20:26 作者：qiya

這篇文章主要為大家介紹了Go并發(fā)編程結(jié)構(gòu)體多字段原子操作示例詳解,有需要的朋友可以借鑒參考下,希望能夠有所幫助,祝大家多多進(jìn)步,早日升職加薪

多字段更新？

并發(fā)編程中，原子更新多個(gè)字段是常見(jiàn)的需求。

舉個(gè)例子，有一個(gè) struct Person 的結(jié)構(gòu)體，里面有兩個(gè)字段。我們先更新 Person.name，再更新 Person.age ，這是兩個(gè)步驟，但我們必須保證原子性。

有童鞋可能奇怪了，為什么要保證原子性？

我們以一個(gè)示例程序開(kāi)端，公用內(nèi)存簡(jiǎn)化成一個(gè)全局變量，開(kāi) 10 個(gè)并發(fā)協(xié)程去更新。你猜最后的結(jié)果是啥？

package main
import (
    "fmt"
    "sync"
    "time"
)
type Person struct {
    name string
    age  int
}
// 全局變量（簡(jiǎn)單處理）
var p Person
func update(name string, age int) {
    // 更新第一個(gè)字段
    p.name = name
    // 加點(diǎn)隨機(jī)性
    time.Sleep(time.Millisecond*200)
    // 更新第二個(gè)字段
    p.age = age
}
func main() {
    wg := sync.WaitGroup{}
    wg.Add(10)
    // 10 個(gè)協(xié)程并發(fā)更新
    for i := 0; i < 10; i++ {
        name, age := fmt.Sprintf("nobody:%v", i), i
        go func() {
            defer wg.Done()
            update(name, age)
        }()
    }
    wg.Wait()
    // 結(jié)果是啥？你能猜到嗎？
    fmt.Printf("p.name=%s\np.age=%v\n", p.name, p.age)
}

打印結(jié)果是啥？你能猜到嗎？

可能是這樣的：

p.name=nobody:2
p.age=3

也可能是：

p.name=nobody:8
p.age=7

按照排列組合來(lái)算，一共有 10*10 種結(jié)果。

那我們想要什么結(jié)果？我們想要 name 和 age 一定要是匹配的，不能牛頭不對(duì)馬嘴。換句話說(shuō)，name 和 age 的更新一定要原子操作，不能出現(xiàn)未定義的狀態(tài)。

我們想要的是（ nobody:i，i ），正確的結(jié)果只能在以下預(yù)定的 10 種結(jié)果出現(xiàn)：

（ nobody:0, 0 ）
（ nobody:1, 1 ）
（ nobody:2, 2 ）
（ nobody:3, 3 ）
...
（ nobody:9, 9 ）

這僅僅是一個(gè)簡(jiǎn)單的示例，童鞋們思考下自己現(xiàn)實(shí)的需求，應(yīng)該是非常常見(jiàn)的。

現(xiàn)在有兩個(gè)問(wèn)題：

第一個(gè)問(wèn)題：這個(gè) demo 觀察下運(yùn)行時(shí)間，用 time 來(lái)觀察，時(shí)間大概是 200 ms 左右，為什么？

root@ubuntu:~/code/gopher/src/atomic_test# time ./atomic_test 
p.name=nobody:8
p.age=7

real 0m0.203s
user 0m0.000s
sys 0m0.000s

如上就是 203 毫秒。劃重點(diǎn)：這個(gè)時(shí)間大家請(qǐng)先記住了，對(duì)我們分析下面的例子有幫助。

這個(gè) 200 毫秒是因?yàn)槠尕笤?nbsp;update 函數(shù)中故意加入了一點(diǎn)點(diǎn)時(shí)延，這樣可以讓程序估計(jì)跑慢一點(diǎn)。

每個(gè)協(xié)程跑 update 的時(shí)候至少需要 200 毫秒，10 個(gè)協(xié)程并發(fā)跑，沒(méi)有任何互斥，時(shí)間重疊，所以整個(gè)程序的時(shí)間也是差不都 200 毫秒左右。

第二個(gè)問(wèn)題：怎么解決這個(gè)正確性的問(wèn)題。

大概兩個(gè)辦法：

鎖互斥
原子操作

下面詳細(xì)分析下異同和優(yōu)劣。

鎖實(shí)現(xiàn)

在并發(fā)的上下文，用鎖來(lái)互斥，這是最常見(jiàn)的思路。鎖能形成一個(gè)臨界區(qū)，鎖內(nèi)的一系列操作任何時(shí)刻都只會(huì)有一個(gè)人更新，如此就能確保更新不會(huì)混亂，從而保證多步操作的原子性。

首先配合變量，對(duì)應(yīng)一把互斥鎖：

// 全局變量（簡(jiǎn)單處理）
var p Person
// 互斥鎖，保護(hù)變量更新
var mu sync.Mutex

更新的邏輯在鎖內(nèi)：

func update(name string, age int) {
    // 更新：加鎖，邏輯串行化
    mu.Lock()
    defer mu.Unlock()

    // 以下邏輯不變
}

大家按照上面的把程序改了之后，邏輯是不是就正確了。一定是（ nobody:i，i ）配套更新的。

但你注意到另一個(gè)可怕的問(wèn)題嗎？

程序運(yùn)行變的好慢?。。?！

同樣用 time 命令統(tǒng)計(jì)下程序運(yùn)行時(shí)間，竟然耗費(fèi) 2 秒?。?！，10 倍的時(shí)延增長(zhǎng)，每次都是這樣。

root@ubuntu:~/code/gopher/src/atomic_test# time ./atomic_test 
p.name=nobody:8
p.age=8

real 0m2.017s
user 0m0.000s
sys 0m0.000s

不禁要問(wèn)自己，為啥？

還記得上面我提到過(guò)，一個(gè) update 固定要 200 毫秒。

加鎖之后的 update 函數(shù)邏輯全部在鎖內(nèi)，10 個(gè)協(xié)程并發(fā)跑 update 函數(shù)，但由于鎖的互斥性，搶鎖不到就阻塞等待，保證 update 內(nèi)部邏輯的串行化。

第 1 個(gè)協(xié)程加上鎖了，后面 9 個(gè)都要等待，依次類推。最長(zhǎng)的等待時(shí)間應(yīng)該是 1.8 秒。

換句話說(shuō)，程序串行執(zhí)行了 10 次 update 函數(shù)，時(shí)間是累加的。程序 2 秒的運(yùn)行時(shí)延就這樣來(lái)的。

加鎖不怕，搶鎖等待才可怕。在大量并發(fā)的時(shí)候，由于鎖的互斥特性，這里的性能可能堪憂。

還有就是搶鎖失敗的話，是要把調(diào)度權(quán)讓出去的，直到下一次被喚醒。這里還增加了協(xié)程調(diào)度的開(kāi)銷(xiāo)，一來(lái)一回可能性能就更慢了下來(lái)。

思考：用鎖之后正確性是保證了，某些場(chǎng)景性能可能堪憂。那咋吧？

在本次的例子，下一步的進(jìn)化就是：原子化操作。

溫馨提示：

怕童鞋誤會(huì)，聲明一下：鎖不是不能用，是要區(qū)分場(chǎng)景，不分場(chǎng)景的性能優(yōu)化措施是沒(méi)有意義的哈。大部分的場(chǎng)景，用鎖沒(méi)啥問(wèn)題。且鎖是可以細(xì)化的，比如讀鎖和寫(xiě)鎖，更新加寫(xiě)鎖，只讀操作加讀鎖。這樣確實(shí)能帶來(lái)較大的性能提升，特別是在寫(xiě)少讀多的時(shí)候。

原子操作

其實(shí)我們?cè)偕罹肯?，這里本質(zhì)上是想要保證更新 name 和 age 的原子性，要保證他們配套。其實(shí)可以先再局部環(huán)境設(shè)置好 Person 結(jié)構(gòu)體，然后一把原子賦值給全局變量即可。Go 提供了 atomic.Value 這個(gè)類型。

怎么改造？

首先把并發(fā)更新的目標(biāo)設(shè)置為 atomic.Value 類型：

// 全局變量（簡(jiǎn)單處理）
var p atomic.Value

然后 update 函數(shù)改造成先局部構(gòu)造，再原子賦值的方式：

func update(name string, age int) {
    lp := &Person{}
    // 更新第一個(gè)字段
    lp.name = name
    // 加點(diǎn)隨機(jī)性
    time.Sleep(time.Millisecond * 200)
    // 更新第二個(gè)字段
    lp.age = age
    // 原子設(shè)置到全局變量
    p.Store(lp)
}

最后 main 函數(shù)讀取全局變量打印的地方，需要使用原子 Load 方式：

    // 結(jié)果是啥？你能猜到嗎？
    _p := p.Load().(*Person)
    fmt.Printf("p.name=%s\np.age=%v\n", _p.name, _p.age)

這樣就解決并發(fā)更新的正確性問(wèn)題啦。感興趣的童鞋可以運(yùn)行下，結(jié)果都是正確的（ nobody:i，i ）。

下面再看一下程序的運(yùn)行時(shí)間：

root@ubuntu:~/code/gopher/src/atomic_test# time ./atomic_test 
p.name=nobody:7
p.age=7

real 0m0.202s
user 0m0.000s
sys 0m0.000s

竟然是 200 毫秒作用，比鎖的實(shí)現(xiàn)時(shí)延少 10 倍，并且保證了正確性。

為什么會(huì)這樣？

因?yàn)檫@ 10 個(gè)協(xié)程還是并發(fā)的，沒(méi)有類似于鎖阻塞等待的操作，只有最后 p.Store(lp) 調(diào)用內(nèi)才有做狀態(tài)的同步，而這個(gè)時(shí)間微乎其微，所以 10 個(gè)協(xié)程的運(yùn)行時(shí)間是重疊起來(lái)的，自然整個(gè)程序就只有 200 毫秒左右。

鎖和原子變量都能保證正確的邏輯。在我們這個(gè)簡(jiǎn)要的場(chǎng)景里，我相信你已經(jīng)感受到性能的差距了。

當(dāng)然了，還是那句話，具體用那個(gè)實(shí)現(xiàn)要看具體場(chǎng)景，不能一概而論。而且，鎖有自己無(wú)可替代的作用，它能保證多個(gè)步驟的原子性，而不僅僅是字段的賦值。

相信你已經(jīng)非常好奇 atomic.Value 了，下面簡(jiǎn)要的分析下原理，是否真的很神秘呢？

原理可能要大跌眼鏡。

趁現(xiàn)在我們還不懂內(nèi)部原理，先思考個(gè)問(wèn)題（不然待會(huì)一下子看懂了就沒(méi)意思了）？

Value.Store 和 Value.Load 是用來(lái)賦值和取值的。我的問(wèn)題是，這兩個(gè)函數(shù)里面有沒(méi)有用戶數(shù)據(jù)拷貝？Store 和 Load 是否是保證了多字段拷貝的原子性？

提前透露下：并非如此。

atomic.Value 原理

atomic.Value 結(jié)構(gòu)體

atomic.Value 定義于文件 src/sync/atomic/value.go ，結(jié)構(gòu)本身非常簡(jiǎn)單，就是一個(gè)空接口：

type Value struct {
    v interface{}
}

在之前文章中，奇伢有分享過(guò) Go 的空接口類型（ interface {} ）在 Go 內(nèi)部實(shí)現(xiàn)是一個(gè)叫做 eface 的結(jié)構(gòu)體（ src/runtime/iface.go ）：

type eface struct {
    _type *_type
    data  unsafe.Pointer
}

interface {} 是給程序猿用的，eface 是 Go 內(nèi)部自己用的，位于不同層面的同一個(gè)東西，這個(gè)請(qǐng)先記住了，因?yàn)?nbsp;atomic.Value 就利用了這個(gè)特性，在 value.go 定義了一個(gè) ifaceWords 的結(jié)構(gòu)體。

劃重點(diǎn)：interface {} ，eface ，ifaceWords 這三個(gè)結(jié)構(gòu)體內(nèi)存布局完全一致，只是用的地方不同而已，本質(zhì)無(wú)差別。這給類型的強(qiáng)制轉(zhuǎn)化創(chuàng)造了前提。

Value.Store 方法

看一下簡(jiǎn)要的代碼，這是一個(gè)簡(jiǎn)單的 for 循環(huán)：

func (v *Value) Store(x interface{}) {
    // 強(qiáng)制轉(zhuǎn)化類型，轉(zhuǎn)變成 ifaceWords （三種類型，相同的內(nèi)存布局，這是前提）
    vp := (*ifaceWords)(unsafe.Pointer(v))
    xp := (*ifaceWords)(unsafe.Pointer(&x))
    for {
        // 獲取數(shù)據(jù)類型
        typ := LoadPointer(&vp.typ)
        // 第一個(gè)判斷：atomic.Value 初始的時(shí)候是 nil 值，那么就是走這里進(jìn)去的；
        if typ == nil {
            runtime_procPin()
            if !CompareAndSwapPointer(&vp.typ, nil, unsafe.Pointer(^uintptr(0))) {
                runtime_procUnpin()
                continue
            }
            // 初始賦值
            StorePointer(&vp.data, xp.data)
            StorePointer(&vp.typ, xp.typ)
            runtime_procUnpin()
            return
        }
        // 第二個(gè)判斷：這個(gè)也是初始的時(shí)候，這是一個(gè)中間狀態(tài)；
        if uintptr(typ) == ^uintptr(0) {
            continue
        }
        // 第三個(gè)判斷：類型校驗(yàn)，通過(guò)這里就能看出來(lái)，Value 里面的類型不能變，否則會(huì) panic；
        if typ != xp.typ {
            panic("sync/atomic: store of inconsistently typed value into Value")
        }
        // 劃重點(diǎn)啦：只要過(guò)了初始化賦值階段，基本上就是直接跑到這行代碼啦
        StorePointer(&vp.data, xp.data)
        return
    }
}

有幾個(gè)點(diǎn)稍微解釋下：

atomic.Value 使用 ^uintptr(0) 作為第一次存取的標(biāo)志位，這個(gè)標(biāo)識(shí)位是設(shè)置在 type 字段里，這是一個(gè)中間狀態(tài)；
通過(guò) CompareAndSwapPointer 來(lái)確保 ^uintptr(0) 只能被一個(gè)執(zhí)行體搶到，其他沒(méi)搶到的走 continue ，再循環(huán)一次；
atomic.Value 第一次寫(xiě)入數(shù)據(jù)時(shí)，將當(dāng)前協(xié)程設(shè)置為不可搶占，當(dāng)存儲(chǔ)完畢后，即可解除不可搶占；
真正的賦值，無(wú)論是第一次，還是后續(xù)的 data 賦值，再 Store 內(nèi)，只涉及到指針的原子操作，不涉及到數(shù)據(jù)拷貝；

這里有沒(méi)有大跌眼鏡？

Store 內(nèi)部并不是保證多字段的原子拷貝！?。?！Store 里面處理的是個(gè)結(jié)構(gòu)體指針。只通過(guò)了 StorePointer 保證了指針的原子賦值操作。

我的天？是這樣的嗎？那何來(lái)的原子操作。

核心在于：Value.Store() 的參數(shù)必須是個(gè)局部變量（或者說(shuō)是一塊全新的內(nèi)存）。

這里就回答了上面的問(wèn)題：Store，Load 是否有數(shù)據(jù)拷貝？

劃重點(diǎn)：沒(méi)有！沒(méi)動(dòng)數(shù)據(jù)

原來(lái)你是這樣子的 atomic.Value ！

回憶一下我上面的 update 函數(shù)，真的是局部變量，全新的內(nèi)存塊：

func update(name string, age int) {
    // 注意哦，局部變量哦
    lp := &amp;Person{}
    // 更新字段 。。。。
 
    // 設(shè)置的是全新的內(nèi)存地址給全局的 atomic.Value 變量
    p.Store(lp)
}

又有個(gè)問(wèn)題，你可能會(huì)想了，如果 p.Store( /* */ ) 傳入的不是指針，而是一個(gè)結(jié)構(gòu)體呢？

事情會(huì)是這樣的：

編譯器識(shí)別到這種情況，編譯期間就會(huì)多生成一段代碼，用 runtime.convT2E 函數(shù)把結(jié)構(gòu)體賦值轉(zhuǎn)化成 eface （注意，這里會(huì)涉及到結(jié)構(gòu)體數(shù)據(jù)的拷貝）；
然后再調(diào)用 Value.Store 方法，所以就 Store 方法而言，行為還是不變；

再思考一個(gè)問(wèn)題：既然是指針的操作，為什么還要有個(gè) for 循環(huán)，還要有個(gè) CompareAndSwapPointer ？

這是因?yàn)?nbsp;ifaceWords 是兩個(gè)字段的結(jié)構(gòu)體，初始賦值的時(shí)候，要賦值類型和數(shù)據(jù)指針兩部分。

atomic.Value 是服務(wù)所有類型，此類需求的，通用封裝。

Value.Load 方法

有寫(xiě)就有讀嘛，看一下讀的簡(jiǎn)要的實(shí)現(xiàn)：

func (v *Value) Load() (x interface{}) {
    vp := (*ifaceWords)(unsafe.Pointer(v))
    typ := LoadPointer(&amp;vp.typ)
    // 初始賦值還未完成
    if typ == nil || uintptr(typ) == ^uintptr(0) {
        return nil
    }
    // 劃重點(diǎn)啦：只要過(guò)了初始化賦值階段，原子讀的時(shí)候基本上就直接跑到這行代碼啦；
    data := LoadPointer(&amp;vp.data)
    xp := (*ifaceWords)(unsafe.Pointer(&amp;x))
    // 賦值類型，和數(shù)據(jù)結(jié)構(gòu)體的地址
    xp.typ = typ
    xp.data = data
    return
}

哇，太簡(jiǎn)單了。處理做了一下初始賦值的判斷（返回 nil ），后續(xù)基本就只靠 LoadPointer 函數(shù)來(lái)個(gè)原子讀指針值而已。

總結(jié)

interface {} ，eface ，ifaceWords 本質(zhì)是一個(gè)東西，同一種內(nèi)存的三種類型解釋，用在不同層面和場(chǎng)景。它們可以通過(guò)強(qiáng)制類型轉(zhuǎn)化進(jìn)行切換；
atomic.Value 使用 cas 操作只在初始賦值的時(shí)候，一旦賦值過(guò)，后續(xù)賦值的原子操作更簡(jiǎn)單，依賴于 StorePointer ，指針值得原子賦值；
atomic.Value 的 Store 和 Load 方法都不涉及到數(shù)據(jù)拷貝，只涉及到指針操作；
atomic.Value 的神奇的核心在于：每次 Store 的時(shí)候用的是全新的內(nèi)存塊！??！且 Load 和 Store 都是以完整結(jié)構(gòu)體的地址進(jìn)行操作，所以才有原子操作的效果。
atomic.Value 實(shí)現(xiàn)多字段原子賦值的原理千萬(wàn)不要以為是并發(fā)操作同一塊多字段內(nèi)存，還能保證原子性；