Golang 并发编程(九)：Golang 中原子操作

Golang 中原子操作

原子操作

原子操作即操作过程中不能被中断的操作。在针对某个值的原子操作执行过程当中，CPU 绝对不会再去执行其他针对该值的操作，无论这些其他操作是否为原子操作。

Golang 语言提供的原子操作都是非侵入式的，他们由标准库代码包 sync.atomic 中的众多函数代表，可以通过这些函数对几种简单类型的值执行原子操作。这些类型包括 6 中：int32、int64、uint32、uint64、uintptr 和 unsafe.Pointer 。这些函数提供的原子操作共有 5 种：增或减、比较并交换、载入、存储和交换。它们分别提供了不同的功能，且适用场景也有所区别。

原子增或减

用于增或减的原子操作的函数名都是以 “Add” 为前缀，后面跟着具体的类型，即 int32、int64、uint32、uint64、uintptr 和 unsafe.Pointer 类型。atomic.AddInt32 要求第一参数必须为指针类型的值，是因为该函数需要获得被操作值在内存中存放的位置，以便施加特殊的 CPU 指令。第二个参数的类型必须与被操作的类型相同。

比如 int32 原子加或减操作：

var i32 int32
i32 = 32
atomic.AddInt32(&i32, 1) // AddInt64
fmt.Println(i32)
atomic.AddInt32(&i32, -2)
fmt.Println(i32)

会输出：

33
31

如果想对 uint32 和 uint64 原子减法，就不能这么直接了，因为它们的第二个参数的类型分别是uint32和uint64，都是无符号的，不过，这也是可以做到的，就是稍微麻烦一些。

如果想对uint32类型的被操作值 32 做原子减法，比如说差量是 -3，那么我们可以先把这个差量转换为有符号的int32类型的值，然后再把该值的类型转换为uint32，用表达式来描述就是 uint32(int32(-3))。先把 int32(-3) 的结果值赋给变量 delta，再把 delta 的值转换为 uint32类型的值，就可以绕过编译器的检查并得到正确的结果了。

var ui32 uint32 = 32
delta := int32(-3)
atomic.AddUint32(&ui32, uint32(delta))
fmt.Println(ui32)

会输出：

还有一种更加直接的方式。我们可以依据下面这个表达式来给定 atomic.AddUint32 函数的第二个参数值：

var ui32 uint32 = 32
atomic.AddUint32(&ui32, ^uint32(-(-3)-1)) //^uint32(-N-1))
fmt.Println(ui32)

其中的 N 代表由负整数表示的差量。也就是说，我们先要把差量的绝对值减去 1，然后再把得到的这个无类型的整数常量，转换为 uint32 类型的值，最后，在这个值之上做按位异或操作，就可以获得最终的参数值了。

此表达式的结果值的补码，与使用前一种方法得到的值的补码相同，所以这两种方式是等价的。我们都知道，整数在计算机中是以补码的形式存在的，所以在这里，结果值的补码相同就意味着表达式的等价。

比较并交换

比较交换即 “Compare and Swap” ，简称CAS。在 sync.atomic 包中，这类原子操作由名称以 “CompareAndSwap” 为前缀的若干函数为代表。

func CompareAndSwapInt32(addr *int32, old, new int32) (swapped bool)

CompareAndSwapInt32 接收三个参数。第一个被操作值的指针值，后面两个参数的类型都是 int32 类型，分别代表操作值的旧值和新。CompareAndSwapInt32 调用后会判断第一个参数 addr 指向的操作值与参数 old 的值是否相等，只有相等时才会使用第三个参数 new 替换，否则忽略掉，并返回false，只有在条件满足的情况下才会进行值的交换。

var num int32 = 32
atomic.CompareAndSwapInt32(&num, 32, 1)
fmt.Println(num)

此例子中会输出：

CAS 操作的优势是：可以在不创建互斥量和不形成临界区的情况下，完成并发安全的值替换操作。这可以大大减少同步对程序性能的损耗。可以看到，CAS操作并不是单一的操作，而是一种操作组合。这与其他的原子操作都不同。正因为如此，它的用途要更广泛一些。例如，我们将它与for 语句联用就可以实现一种简易的自旋锁（spinlock）。

sign := make(chan struct{}, 2)
num := int32(0)
fmt.Printf("The number: %d\n", num)
go func() { // 定时增加num的值。
    defer func() {
       sign <- struct{}{}
    }()
    for {
       time.Sleep(time.Millisecond * 500)
       newNum := atomic.AddInt32(&num, 2)
       fmt.Printf("The number: %d\n", newNum)
       if newNum == 10 {
          break
       }
    }
}()
go func() { // 定时检查num的值，如果等于10就将其归零。
    defer func() {
       sign <- struct{}{}
    }()
    for {
       if atomic.CompareAndSwapInt32(&num, 10, 0) {
          fmt.Println("The number has gone to zero.")
          break
       }
       time.Sleep(time.Millisecond * 500)
    }
}()
<-sign
<-sign

在 for 语句中的CAS操作可以不停地检查某个需要满足的条件，一旦条件满足就退出 for 循环。这就相当于，只要条件未被满足，当前的流程就会被一直“阻塞”在这里。

注意：在 for 语句加CAS操作的假设往往是：共享资源状态的改变并不频繁，或者，它的状态总会变成期望的那样。这是一种更加乐观，或者说更加宽松的做法。

载入

前面对增加或减以及交换 都是原子操作，那对它进行读操作的时候，还有必要使用原子操作吗？有必要，在 32 位计算机架构中写入一个 64 位的整数。如果在这个写操作完成前，有一个读操作并发地进行了，这个读操作就可能读取到一个只能被修改了一半的值。这种结果是非常糟糕的。

所以，一旦决定了要对一个共享资源进行保护，那就要做到完全的保护。不完全的保护基本上与不保护没有什么区别。

for {
    v := atomic.LoadInt32(&num)
    if atomic.CompareAndSwapInt32(&num, v, 0) {
        fmt.Println("The number has gone to zero.")
        break
    }
    time.Sleep(time.Millisecond * 500)
}

这里原地读取变量 num 的值并把它赋给变量 v，这样一来，读取 value 的值时，这样当前计算机中任何 CPU 都不会进行其他针对此值的读写操作。虽然这里使用了 atomic.LoadInt32 函数原子地载入 value 的值，但是其他后面的 CAS 任然是有必要的。因为，那条赋值语句和后面的 if 语句并不会原子地执行。

存储

与读取操作相对应的是写入操作。而 sync.atomic 包也提供了对应的存储函数，这些函数的名称均以 “store” 为前缀。

在原子存储某个值的过程中，任何 CPU 都不会进行对同一个值的读写操作。如果把所有针对此值的写操作都改为原子操作，就绝对不会出现针对此值的读操作因被并发地进行，而读到修改了一半的值的情况。

func StoreInt32(addr *int32, val int32)

函数 StoreInt32 会接收两个参数，第一个参数的类型是 *int32 ，它同样是指向被操作的指针值。而第二个参数则是 int32 类型的，它的值是欲存储的新值。

注意：原子的值存储操作总会成功，因为它并不关心被操作值的旧值是什么。

原子值

为了扩大原子操作的适用范围，Golang 语言在 1.4 版本发布的时候向 sync/atomic 包中添加了一个新的类型 Value。此类型的值相当于一个容器，可以被用来 “原子地” 存储和加载任意的值。

我们先举例一个编发读写的例子：假设我们现在仓库中有 32 个库存数的商品，我们生产每次会以 +3 个的形式出现，卖出则会以 -2 的形式出现，其中卖出和生产都需要一定的处理时间，这里我使用 time.Sleep(time.Microsecond * 500) 模拟耗时。

func main() {
    var num int32 = 32 // 库存数
    total := 10 // 总并发数
    var wg sync.WaitGroup
    wg.Add(total * 2)
    for i := 0; i < total; i++ {
        go func() {
            fmt.Printf("进行了 +3 操作，num=%d\n", num)
            num = add(num)
            wg.Done()
        }()
    }
    for i := 0; i < total; i++ {
        go func() {
            fmt.Printf("进行了 -2 操作，num=%d\n", num)
            num = sub(num)
            wg.Done()
        }()
    }
    wg.Wait()
    fmt.Println(num)
}

func add(v int32) int32 {
    time.Sleep(time.Microsecond * 500) // 模拟异步逻辑处理操作
    return v + 3
}

func sub(v int32) int32 {
    time.Sleep(time.Microsecond * 500) // 模拟异步逻辑处理操作
    return v - 2
}

执行代码后，你会发现：

进行了 +3 操作，num=32
进行了 -2 操作，num=32
....
进行了 -2 操作，num=32
进行了 +3 操作，num=32
35

数据永远不对，这是因为进行并发读取，生产的时候又消费，导致库存不准确。我们可以使用最简单的互斥锁，生产的时候只能生产，消费的时候只能消费。

func main() {
    var num int32 = 32
    total := 10 // 总并发数
    var mutex sync.Mutex
    var wg sync.WaitGroup
    wg.Add(total * 2)
    for i := 0; i < total; i++ {
        go func() {
            fmt.Printf("进行了 +3 操作，num=%d\n", num)
            mutex.Lock()
            defer mutex.Unlock()
            num = add(num)
            wg.Done()
        }()
    }
    for i := 0; i < total; i++ {
        go func() {
            fmt.Printf("进行了 -2 操作，num=%d\n", num)
            mutex.Lock()
            defer mutex.Unlock()
            num = sub(num)
            wg.Done()
        }()
    }
    wg.Wait()
    fmt.Println(num)
}

func add(v int32) int32 {
    time.Sleep(time.Microsecond * 500) // 模拟异步逻辑处理操作
    return v + 3
}

func sub(v int32) int32 {
    time.Sleep(time.Microsecond * 500) // 模拟异步逻辑处理操作
    return v - 2
}

这次我们发现现在无论如何生产消费，库存总是正确的，不会发生什么改变，于是我们想了想，这种直接锁住当前生产者和消费者的损耗是不是太大了？于是我们可以考虑用原子操作试着实现。

func main() {
    var num int32 = 32
    total := 10 // 总并发数
    var wg sync.WaitGroup
    wg.Add(total * 2)
    for i := 0; i < total; i++ {
        go func() {
            fmt.Printf("进行了 +3 操作，num=%d\n", num)
            func() {
                time.Sleep(time.Microsecond * 500)
                atomic.AddInt32(&num, 3)
            }()
            wg.Done()
        }()
    }
    for i := 0; i < total; i++ {
        go func() {
            fmt.Printf("进行了 -2 操作，num=%d\n", num)
            func() {
                time.Sleep(time.Microsecond * 500)
                atomic.AddInt32(&num, -2)
            }()
            wg.Done()
        }()
    }
    wg.Wait()
    fmt.Println(num)
}