Go36-29,30-原子操作

原子操作

對(duì)于一個(gè)Go程序來(lái)說(shuō)，GO語(yǔ)言運(yùn)行時(shí)系統(tǒng)中的調(diào)度器會(huì)恰當(dāng)?shù)陌才牌渲兴械膅oroutine的運(yùn)行。不過(guò)，在同一時(shí)刻，只會(huì)有少數(shù)的goroutine真正處于運(yùn)行狀態(tài)。為了公平起見(jiàn)，調(diào)度器會(huì)頻繁的切換這些goroutine。這個(gè)中斷的時(shí)機(jī)有很多，任何兩個(gè)語(yǔ)句執(zhí)行的間隙，甚至是在某條語(yǔ)句執(zhí)行的過(guò)程中都是可能的。即使這些語(yǔ)句在臨界區(qū)之內(nèi)也是一樣的，互斥鎖雖然能保護(hù)臨界區(qū)中的代碼串行執(zhí)行，但是不能保證這些代碼的原子性（atomicity）。

原子操作的特點(diǎn)

真正能夠保證原子性執(zhí)行的工具只有原子操作（atomic operation）。
原子操作在進(jìn)行的過(guò)程中是不允許中斷的。在底層，這會(huì)由CPU提供芯片級(jí)別的支持，所以絕對(duì)有效。
原子操作可以完全的消除競(jìng)態(tài)條件，并能夠絕對(duì)的保證并發(fā)安全。并且它的執(zhí)行速度比其他的同步工具快得多，通常會(huì)高出好幾個(gè)數(shù)量級(jí)。
缺點(diǎn)
正是因?yàn)樵硬僮鞑荒鼙恢袛?，所要它需要足夠?jiǎn)單，并且要求快速。因此，操作系統(tǒng)層面值對(duì)針對(duì)二進(jìn)制位或整數(shù)的原子操作提供支持。
Go語(yǔ)言的原子操作是基于CPU和操作系統(tǒng)的，所以它也只針對(duì)少數(shù)數(shù)據(jù)類型的值提供了原子操作函數(shù)。這些函數(shù)都在標(biāo)準(zhǔn)庫(kù)的sync/atomic中。

sync/atomic 包

sync/atomic 包中可以做的原子操作有：

• 加法（add）
• 比較并交換（compare and swap），簡(jiǎn)稱CAS
• 加載（load）
• 存儲(chǔ)（store）
• 交換（swap）

這些函數(shù)針對(duì)的數(shù)據(jù)類型并不多。但是，對(duì)這些類型中的每一個(gè)，sync/stomic包都會(huì)有一套函數(shù)給予支持。這些數(shù)據(jù)類型有：

• int32
• int64
• uint32
• uint64
• uintptr
• unsafe.Pointer，這個(gè)類型在包里沒(méi)有提供原子加法操作的函數(shù)
• atomic.Value，包里還提供了這個(gè)類型，可以被用來(lái)存儲(chǔ)任意類型的值

函數(shù)需要傳入被操作值的指針

原子操作函數(shù)的第一個(gè)參數(shù)值都應(yīng)該是那個(gè)被操作的值，并且是傳入指針，比如：*int32。
原子操作函數(shù)需要的是被操作值的指針，而不是值本身。即使是unsafe.Pointer類型雖然本身已經(jīng)是指針類型，但是原子操作函數(shù)里還要要這個(gè)值的指針。
只要原子操作函數(shù)拿到了被操作值的指針，就可以定位到存儲(chǔ)該值的內(nèi)存地址。只有這樣，才能夠通過(guò)底層的指令，操作這個(gè)內(nèi)存地址上的數(shù)據(jù)。

加法操作也可以用來(lái)做減法

包里只提供了加法操作的函數(shù)，沒(méi)有減法操作的函數(shù)，不過(guò)是可以實(shí)現(xiàn)減法的。比如：atomic.AddInt32。函數(shù)的第二個(gè)參數(shù)代表的差量，它的類型是int32，這個(gè)類型是有符號(hào)的。這里可以傳個(gè)負(fù)整數(shù)就是做減法了。
對(duì)于atomic.AddInt64也是類型的。不過(guò)對(duì)于atomic.AddUint32和atomic.AddUint64要做原子減法就不能這么直接了，因?yàn)榈诙?shù)的值是uint32和uint64，這些類型是無(wú)符號(hào)的。比如要減3，差量就是-3，要先把差量轉(zhuǎn)換為有符號(hào)的類型比如int32，然后再把該值轉(zhuǎn)換為uint32，用表達(dá)式描述就是：

uint32(int32(-3))

不過(guò)上面這樣寫，會(huì)使編譯器報(bào)錯(cuò)，因?yàn)檫@么做其實(shí)會(huì)讓表達(dá)式的結(jié)果值溢出。不過(guò)可以先把int32(-3)先賦值給一個(gè)臨時(shí)變量比如名字就叫delta，來(lái)繞過(guò)編譯器的檢查。
上面的那種方式比較好理解，另外還有一種方式第二的參數(shù)用下面的表達(dá)式：

^uint32(-(-3)-1)

上面這么做的原理，簡(jiǎn)單來(lái)說(shuō)就是取補(bǔ)碼，具體就要去了解一下計(jì)算機(jī)中的原碼、補(bǔ)碼、反碼，以及是如何實(shí)現(xiàn)減法的了。
上面兩中方式是等價(jià)的，下面是實(shí)例代碼：

package main

import (
    "fmt"
    "sync/atomic"
)

func main() {
    num := uint32(18)
    delta := int32(-3)
    atomic.AddUint32(&num, uint32(delta))
    fmt.Println(num)

    atomic.AddUint32(&num, ^uint32(-(-3)-1))
    fmt.Println(num)
}

比較并交換

比較并交換操作即CAS操作，是有條件的交換操作，只有在條件滿足的情況下才會(huì)進(jìn)行值的交換。
交換，指的是把新值賦值給變量，并返回變量的舊值。
在進(jìn)行CAS操作的時(shí)候，函數(shù)會(huì)先判斷被操作變量的值，是否與預(yù)期的舊值相等。如果相等，就把新值賦給該變量，并返回true以表明交換操作已經(jīng)進(jìn)行。否則就忽略交換操作，并返回false。CAS操作并不是一個(gè)單一的操作，而是一種操作組合。這與其他原子操作都不同。正式因?yàn)槿绱?，它的用途要更廣泛一些。比如，將它與for語(yǔ)句聯(lián)用，就可以實(shí)現(xiàn)一種簡(jiǎn)易的自旋鎖（spinlock）。

自旋鎖
自旋鎖（spinlock）：是指當(dāng)一個(gè)線程在獲取鎖的時(shí)候，如果鎖已經(jīng)被其它線程獲取，那么該線程將循環(huán)等待，然后不斷的判斷鎖是否能夠被成功獲取，直到獲取到鎖才會(huì)退出循環(huán)。

for {
    if atomic.CompareAndSwapInt32(&num, 10, 0) {
        fmt.Println("檢查到num為10，清0")
        break
    }
    time.Sleep(time.Millisecond * 300)
}

下面是代碼完整的展示，實(shí)現(xiàn)了建議的自旋鎖：

package main

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

func main() {
    done := make(chan struct{})
    var num int32

    // 定時(shí)增加num的值
    go func() {
        defer func() {
            done <- struct{}{}
        }()
        for {
            time.Sleep(time.Millisecond * 500)
            newNum := atomic.AddInt32(&num, 1)
            fmt.Println("num:", newNum)
            if newNum > 10 {
                break
            }
        }
    }()

    // 定時(shí)檢查num的值，如果等于10，就設(shè)置為0
    go func() {
        defer func() {
            done <- struct{}{}
        }()
        for {
            if atomic.CompareAndSwapInt32(&num, 10, 0) {
                fmt.Println("檢查到num為10，清0")
                break
            }
            time.Sleep(time.Millisecond * 500)
        }
    }()

    <- done
    <- done
    fmt.Println("Over")
}

樂(lè)觀鎖
在for語(yǔ)句中的CAS操作，可以不停的檢查某個(gè)需要滿足的條件，一旦條件滿足就退出for循環(huán)。這相當(dāng)于只要條件不滿足，當(dāng)前流程就會(huì)被一直阻塞在這里。
這個(gè)在效果上與互斥鎖類型，不過(guò)適用場(chǎng)景不同?；コ怄i總是假設(shè)共享資源的狀態(tài)會(huì)被其他的goroutine頻繁的改變，是一種悲觀鎖。而這里是假設(shè)共享資源狀態(tài)的改變并不頻繁，所以你的操作一般都是如期望的那樣成功，是一種更加樂(lè)觀，更加寬松的做法，就是樂(lè)觀鎖。
下面的例子，啟用了多個(gè)goroutine都要對(duì)num的值做加法操作。

package main

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

var done chan struct{} = make(chan struct{})
var num int32

func CompareAndAdd(id, times int, increment int32) {
    defer func() {
        done <- struct{}{}
    }()
    for i := 0; i < times; i++ {
        for {
            currNum := atomic.LoadInt32(&num)  // 先獲取當(dāng)前的值
            newNum := currNum + increment  // 這里希望對(duì)num做加法
            // 假設(shè)是一個(gè)耗時(shí)的操作，就是有可能在這段時(shí)間里有別的goroutine已經(jīng)修改了num
            time.Sleep(time.Millisecond * 300)
            // 比較現(xiàn)在num的值和操作前獲取到的值是否一致，如果一致，表示表里沒(méi)有別修改過(guò)，可以更新為新值
            // 如果不一致，表示在這段時(shí)間里num已經(jīng)被別的goroutine修改過(guò)了，必須重新來(lái)過(guò)
            if atomic.CompareAndSwapInt32(&num, currNum, newNum) {
                fmt.Printf("更新num[%d-%d]: +%d = %d\n", id, i, increment, newNum)
                break
            } else {
                fmt.Printf("更新num失敗[%d-%d]，重試...\n", id, i)
            }
        }
    }
}

func main() {
    go CompareAndAdd(1, 6, 2)
    go CompareAndAdd(2, 4, 3)
    go CompareAndAdd(3, 3, 4)
    <- done
    <- done
    <- done
    fmt.Println("Over")
}

這里在把加法后的新值賦值給原來(lái)的變量num前，先檢查此時(shí)num的值是否發(fā)生過(guò)變化了，如果沒(méi)有發(fā)生變化，就可以將num設(shè)置為新值。否則就從頭在做一次加法運(yùn)行、檢查、賦值，直到成功為止。在假設(shè)共享資源狀態(tài)的改變并不頻繁的前提下，這種實(shí)現(xiàn)是比悲觀鎖更好的。

讀寫操作都要實(shí)現(xiàn)原子操作

在已經(jīng)保證了對(duì)一個(gè)變量的寫操作都是原子操作，比如：加法、存儲(chǔ)、交換等等。在對(duì)它進(jìn)行讀操作的時(shí)候，依然有必要使用原子操作。
參考讀寫鎖，寫操作和讀操作之間是互斥的，這是為了防止讀操作讀到還沒(méi)有被修改完的值。如果讀操作讀到一半就被中斷了，等再回來(lái)繼續(xù)讀取的時(shí)候，就讀到了修改前后兩部分的內(nèi)容。這顯然破壞了值的完整性。所以，一旦決定對(duì)一個(gè)共享資源進(jìn)行保護(hù)，就要做到完全的保護(hù)。

適用場(chǎng)景

由于原子操作函數(shù)只支持非常有限的數(shù)據(jù)類型，所以在很多應(yīng)用場(chǎng)景下，互斥鎖更加合適。不過(guò)如果當(dāng)前場(chǎng)景下可以使用原子操作，就不要考慮互斥鎖了。
因?yàn)樵硬僮骱瘮?shù)的執(zhí)行速度要比互斥鎖快的多。而且，使用起來(lái)也更加簡(jiǎn)單，不會(huì)涉及臨界區(qū)的選擇，以及死鎖等問(wèn)題。就是原子操作更加高效，而互斥鎖適用場(chǎng)景更廣，優(yōu)先考慮是否可以使用原子操作。

原子變量

為了擴(kuò)大原子操作的適用范圍，Go語(yǔ)言在1.4版本之后，在sync/atomic包中添加了一個(gè)新類型Value。此類型的值相當(dāng)于一個(gè)容器，可以被用來(lái)原子的存儲(chǔ)和加載任意的值。atomic.Value類型是開(kāi)箱即用的，聲明一個(gè)該類型的變量之后就可以直接使用了，可以稱它為原子變量。而原子變量的值，可以稱為原子值。
這個(gè)類型使用起來(lái)很簡(jiǎn)單，只有兩個(gè)指針?lè)椒ǎ篠tore()和Load()，不過(guò)還是有一些需要注意的地方。

原子值的復(fù)制

一旦atomic.Value類型的值，就是原子值被真正使用，它就不應(yīng)該再被復(fù)制了。
只要用它來(lái)存儲(chǔ)值了，就相當(dāng)于開(kāi)始真正使用了。atomic.Value類型屬于結(jié)構(gòu)體類型，而結(jié)構(gòu)體類型屬于值類型。
所以，復(fù)制該類型的值會(huì)產(chǎn)生一個(gè)完全分離的新值。這個(gè)新值相當(dāng)于被復(fù)制的那個(gè)值的一個(gè)快照。之后，不論后者存儲(chǔ)的值怎樣改變，都不會(huì)影響到前者的使用，反之亦然。

這個(gè)是進(jìn)行驗(yàn)證的示例代碼：

func main() {
    var box atomic.Value
    box2 := box  // 原子值真正使用之前可以被復(fù)制
    v1 := [...]int{1,2,3}
    box.Store(v1)  // 對(duì)box1的改變，不會(huì)影響到box2
    fmt.Println(box.Load())
    fmt.Println(box2.Load())
}

上面我把原話都引用過(guò)來(lái)了，下面是我的理解。上面的box2 := box這句，編譯器是有綠色的提示的，不影響運(yùn)行但是應(yīng)該要引起我們注意。然后在源碼里也找到了一些建議：

// A Value provides an atomic load and store of a consistently typed value.
// The zero value for a Value returns nil from Load.
// Once Store has been called, a Value must not be copied.
//
// A Value must not be copied after first use.
type Value struct {
    noCopy noCopy

    v interface{}
}

// noCopy may be embedded into structs which must not be copied
// after the first use.
//
// See https://github.com/golang/go/issues/8005#issuecomment-190753527
// for details.
type noCopy struct{}

看著意思也就是上面說(shuō)的，使用之后就不要再?gòu)?fù)制了。
不過(guò)既然這個(gè)類型是開(kāi)箱即用的，那么只要再聲明一個(gè)變量使用就好了，沒(méi)有必要復(fù)制一個(gè)來(lái)使用。另外就是因?yàn)檫@是一個(gè)值類型，所以賦值的是副本，對(duì)原值的改變不會(huì)影響到副本。如果需要，就用指針。
上面示例中復(fù)制的用法很傻，應(yīng)該不會(huì)也想不到要這么用。仔細(xì)想想，真正需要復(fù)制該值的情況可能是作為函數(shù)的參數(shù)，就是先聲明一個(gè)開(kāi)箱即用的原子值，然后在不同的函數(shù)里都把這個(gè)原子值作為參數(shù)。這里是可以的，而且也很方便。只要知道每個(gè)函數(shù)里都是不同的原子值就行了，就是值類型傳參要注意的那些問(wèn)題。下面是我想的一個(gè)場(chǎng)景：

package main

import (
    "fmt"
    "sync/atomic"
)

func loadBox(box atomic.Value, v interface{}) {
    box.Store(v)
    fmt.Println(box.Load())
}

func main() {
    var box atomic.Value  // 下面調(diào)用了3次函數(shù)，就是復(fù)制了box3次
    v1 := [...]int{1,2,3}
    loadBox(box, v1)
    v2 := "Hello"
    loadBox(box, v2)
    v3 := 123
    loadBox(box, v3)
}

原子值儲(chǔ)值的規(guī)則

用原子值來(lái)儲(chǔ)值有兩條強(qiáng)制性的規(guī)則：

1. 不能用原子值存儲(chǔ)nil
2. 向原子值存儲(chǔ)額第一個(gè)值，決定了它今后能且只能存儲(chǔ)哪一個(gè)類型的值

不能存儲(chǔ)nil
就是不能把nil作為參數(shù)傳入原子值的Store方法，否則就會(huì)引發(fā)panic。
這里還有注意接口類型的變量，它的動(dòng)態(tài)值是nil，但是動(dòng)態(tài)類型卻不是nil，所以它的值就不等于nil。這樣的一個(gè)變量的值是可以被存入原子值的。
就是不能存nil，box.Store(nil)這樣是要panic的。只要是有類型的，值是nil也是可以的，下面這樣用是沒(méi)問(wèn)題的：

func main() {
    var box atomic.Value
    var v1 chan struct{}
    fmt.Println(v1)
    box.Store(v1)
    fmt.Println(box.Load())
}

上面提到了接口，其實(shí)Stroe方法接收的參數(shù)就是空接口：

func (v *Value) Store(x interface{}) {
    // 省略函數(shù)內(nèi)容
}

存儲(chǔ)的類型
接著上面的說(shuō)，Store接收的參數(shù)是一個(gè)空接口，并且還說(shuō)了不能是nil。所以只要是不是nil都可以作為參數(shù)。這只是作為第一次使用的情況。
一旦向原子值存儲(chǔ)了第一個(gè)值，就決定了類型，之后再要存儲(chǔ)，就必須還是同樣的類型了。這個(gè)規(guī)則，就是通過(guò)接口也是繞不開(kāi)的。原子值內(nèi)部是依據(jù)被存儲(chǔ)值的實(shí)際類型來(lái)做判斷的。
這里還有個(gè)問(wèn)題，我們是無(wú)法通過(guò)某個(gè)方法獲知一個(gè)原子值是否已經(jīng)被真正使用。并且，也沒(méi)有辦法通過(guò)常規(guī)的途徑得到一個(gè)原子值可以存儲(chǔ)值的實(shí)際類型。這使得誤用原子值的可能性大大增加，尤其是在多個(gè)地方使用同一個(gè)原子值的時(shí)候。
通過(guò)下面的示例，可以理解一下：

func main() {
    var box atomic.Value
    box.Store("")  // 存入字符串
    box2 := box  // 在真正使用之后，就不應(yīng)該被復(fù)制
    // box2.Store(1)  // 存字符串以外的類型就會(huì)引發(fā)panic
    box2.Store("1")
    _ = box2
}

使用建議

一、不要把內(nèi)部使用的原子值暴露給外界。比如，聲明一個(gè)全局的原子變量并不是一個(gè)正確的做法。這個(gè)變量的訪問(wèn)權(quán)限最起碼也應(yīng)該是包級(jí)私有的。
二、如果不得不讓包外，或者模塊外的代碼使用你的原子值，那么可以聲明一個(gè)包級(jí)私有的原子變量，然后再通過(guò)一個(gè)或多個(gè)公開(kāi)的函數(shù)，讓外界間接的使用到它。注意，這種情況下不要把原子值傳遞到外界，不論是傳遞原子值本身還是它的指針。
三、如果通過(guò)某個(gè)函數(shù)可以向內(nèi)部的原子值存儲(chǔ)值的話，那么就應(yīng)該在這個(gè)函數(shù)中先判斷被存儲(chǔ)值類型的合法性。若不合法，則應(yīng)該直接返回對(duì)應(yīng)的錯(cuò)誤值，從而避免panic的發(fā)生。
四、如果可能的話，我們可以把原子值封裝到一個(gè)數(shù)據(jù)類型中，比如一個(gè)結(jié)構(gòu)體。這樣我們既可以通過(guò)該類型的方法更加安全地存儲(chǔ)值，又可以在該類型中包含可村儲(chǔ)值的合法類型信息。

概括一下，上面說(shuō)的就是一個(gè)最佳實(shí)踐，用一個(gè)結(jié)構(gòu)體來(lái)封裝。并且解決了前面提到的沒(méi)有辦法獲取到原子值存儲(chǔ)的實(shí)際類型的問(wèn)題：

package main

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

// 創(chuàng)建結(jié)構(gòu)體，封裝atomic.Value和村儲(chǔ)值的合法類型
// 字段都是私有的，下面提供了4個(gè)可導(dǎo)出的方法
type atomicValue struct {
    v atomic.Value
    t reflect.Type
}

// 提供方法，返回存儲(chǔ)值的合法類型
func (av *atomicValue) TypeOfValue() reflect.Type {
    return av.t
}

// 提供方法，存儲(chǔ)值。存儲(chǔ)之前先檢查類型
func (av *atomicValue) Store(v interface{}) error {
    if v == nil {
        return fmt.Errorf("不能存儲(chǔ)nil")
    }
    t := reflect.TypeOf(v)
    if t != av.t {
        return fmt.Errorf("類型不正確, 需要: %s, 實(shí)際: %s", av.t, t)
    }
    av.v.Store(v)
    return nil
}

// 提供方法，獲取值，雖然示例中沒(méi)有用到
func (av *atomicValue) Load() interface{} {
    return av.v.Load()
}

// 創(chuàng)建結(jié)構(gòu)體的方法，相當(dāng)于構(gòu)造方法
func NewAtomicValue(x interface{}) (*atomicValue, error) {
    if x == nil {
        return nil, fmt.Errorf("不能存儲(chǔ)nil")
    }
    return &atomicValue{
        t: reflect.TypeOf(x),  // 獲取變量的類型，返回reflect.Type類型
    }, nil
}

func main() {
    v := fmt.Errorf("隨便的錯(cuò)誤")
    box, err := NewAtomicValue(v)
    if err != nil {
        fmt.Fprintf(os.Stderr, "ERROR: %s\n", err)
    }
    fmt.Printf("合法的類型是: %s\n", box.TypeOfValue())
    v2 := fmt.Errorf("還是一個(gè)錯(cuò)誤類型")
    err = box.Store(v2)
    if err != nil {
        fmt.Fprintf(os.Stderr, "ERROR: %s\n", err)
    }
    fmt.Printf("存儲(chǔ)了一個(gè)值，類型是: %T\n", v2)
    fmt.Println("嘗試存儲(chǔ)一個(gè)其他類型的值")
    err = box.Store(1)
    if err != nil {
        fmt.Fprintf(os.Stderr, "ERROR: %s\n", err)
    }
}

存儲(chǔ)引用類型

這里還要特別強(qiáng)調(diào)一點(diǎn)：盡量不要向原子值中存儲(chǔ)引用類型的值。因?yàn)檫@很容易造成安全漏洞。盡量不要的意思就是要存還是可以存的，下面的示例中也給出了建議的方法：

package main

import (
    "fmt"
    "sync/atomic"
)

func main() {
    var box atomic.Value
    v := []int{1,2,3}  // 切片是引用類型
    box.Store(v)
    v[1] = 4  // 此處的操作不是并發(fā)安全的！
    fmt.Println(box.Load())  // 存儲(chǔ)的值被改變了

    // 正確的做法如下：
    // 下面這個(gè)函數(shù)就是把引用類型復(fù)制一份出來(lái)，然后存儲(chǔ)起來(lái)
    // 類似于把一個(gè)值類型傳遞給函數(shù)的效果
    store := func(v []int) {
        replica := make([]int, len(v))
        copy(replica, v)
        box.Store(replica)
    }
    store(v)
    v[2] = 5  // 再試著改變切面的值
    fmt.Println(box.Load())  // 存儲(chǔ)的是副本的值，不會(huì)被上面的改變影響
}

這里把一個(gè)切片類型存儲(chǔ)了原子值。切片類型屬于引用類型，所以在外面依然可以改變切片的值。這相當(dāng)于繞過(guò)了原子值而進(jìn)行了非并發(fā)安全的操作。
這里應(yīng)該先為切片創(chuàng)建一個(gè)完全的副本，然后再把副本存儲(chǔ)box。如此一來(lái)，在對(duì)原來(lái)的切片做修改都不會(huì)破壞box提供的安全保護(hù)。

總結(jié)

原子操作明顯比互斥鎖要更加輕便，但是限制也很明顯。所以如果可以使用原子操作的話，一定是用原子操作更好。
現(xiàn)在有了原子值，突破了一些原子操作的限制。在原子值與互斥鎖之間選擇的時(shí)候，就需要仔細(xì)考慮了。這篇里講了很多使用原子值時(shí)候的注意事項(xiàng)，可能用的時(shí)候就不如互斥鎖這么好用了。
另外在CAS中還會(huì)遇到一個(gè)ABA問(wèn)題，而原子類型應(yīng)該就會(huì)有這個(gè)ABA問(wèn)題。此時(shí)就要用互斥鎖了，除非業(yè)務(wù)對(duì)ABA問(wèn)題不敏感