Golang同步：原子操作使用

• GO语言提供的原子操作都是非入侵式的，由标准库sync/atomic中的众多函数代表
• 类型包括int32,int64,uint32,uint64,uintptr,unsafe.Pointer，共六个。
• 这些函数提供的原子操作共有五种：增或减，比较并交换，载入，存储和交换

int各种类型取值范围

增或减Add

函数名称都以Add为前缀，并后跟针对的具体类型的名称。

• 被操作的类型只能是数值类型
• int32,int64,uint32,uint64,uintptr类型可以使用原子增或减操作
• 第一个参数值必须是一个指针类型的值，以便施加特殊的CPU指令
• 第二个参数值的类型和第一个被操作值的类型总是相同的。

示例

 package main

import (

"fmt"

"sync/atomic"

)

func main(){

var i32 int32

fmt.Println("=====old i32 value=====")

fmt.Println(i32)

//第一个参数值必须是一个指针类型的值,因为该函数需要获得被操作值在内存中的存放位置,以便施加特殊的CPU指令

//结束时会返回原子操作后的新值

newI32 := atomic.AddInt32(&i32,3)

fmt.Println("=====new i32 value=====")

fmt.Println(i32)

fmt.Println(newI32)

var i64 int64

fmt.Println("=====old i64 value=====")

fmt.Println(i64)

newI64 := atomic.AddInt64(&i64,-3)

fmt.Println("=====new i64 value=====")

fmt.Println(i64)

fmt.Println(newI64)

}  

运行结果：

  /usr/local/go/bin/go run /Users/liuxinming/go/src/free/learngo/atomic/add/add.go

=====old i32 value=====

0

=====new i32 value=====

-3

-3

=====old i64 value=====

0

=====new i64 value=====

-3

-3  

比较并交换CAS

Compare And Swap 简称CAS，在sync/atomic包种，这类原子操作由名称以‘CompareAndSwap’为前缀的若干个函数代表。

• 声明如下

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

• 调用函数后，会先判断参数addr指向的被操作值与参数old的值是否相等
• 仅当此判断得到肯定的结果之后，才会用参数new代表的新值替换掉原先的旧值，否则操作就会被忽略。
• so, 需要用for循环不断进行尝试,直到成功为止
• 使用锁的做法趋于悲观

– 我们总假设会有并发的操作要修改被操作的值，并使用锁将相关操作放入临界区中加以保护

• 使用CAS操作的做法趋于乐观

– 总是假设被操作值未曾被改变（即与旧值相等），并一旦确认这个假设的真实性就立即进行值替换。

示例

 package main

import (

"fmt"

"sync/atomic"

)

var value int32

func main() {

fmt.Println("======old value=======")

fmt.Println(value)

fmt.Println("======CAS value=======")

addValue(3)

fmt.Println(value)

}

//不断地尝试原子地更新value的值,直到操作成功为止

func addValue(delta int32){

//在被操作值被频繁变更的情况下,CAS操作并不那么容易成功

//so 不得不利用for循环以进行多次尝试

for {

v := value

if atomic.CompareAndSwapInt32(&value, v, (v + delta)){

//在函数的结果值为true时,退出循环

break

}

//操作失败的缘由总会是value的旧值已不与v的值相等了.

//CAS操作虽然不会让某个Goroutine阻塞在某条语句上,但是仍可能会使流产的执行暂时停一下,不过时间大都极其短暂.

}

}  

结果：

 ======old value=======
0
======CAS value=======
3  

载入Load

上面的比较并交换案例总 v:= value为变量v赋值，但… 要注意，在进行读取value的操作的过程中,其他对此值的读写操作是可以被同时进行的,那么这个读操作很可能会读取到一个只被修改了一半的数据.

* so so so , 我们要使用sync/atomic代码包同样为我们提供了一系列的函数，以Load为前缀(载入)，来确保这样的糟糕事情发生。

示例

 package main

import (

"fmt"

"sync/atomic"

)

var value int32

func main() {

fmt.Println("======old value=======")

fmt.Println(value)

fmt.Println("======CAS value=======")

addValue(3)

fmt.Println(value)

}

//不断地尝试原子地更新value的值,直到操作成功为止

func addValue(delta int32){

//在被操作值被频繁变更的情况下,CAS操作并不那么容易成功

//so 不得不利用for循环以进行多次尝试

for {

//v := value

//在进行读取value的操作的过程中,其他对此值的读写操作是可以被同时进行的,那么这个读操作很可能会读取到一个只被修改了一半的数据.

//因此我们要使用载入

v := atomic.LoadInt32(&value)

if atomic.CompareAndSwapInt32(&value, v, (v + delta)){

//在函数的结果值为true时,退出循环

break

}

//操作失败的缘由总会是value的旧值已不与v的值相等了.

//CAS操作虽然不会让某个Goroutine阻塞在某条语句上,但是仍可能会使流产的执行暂时停一下,不过时间大都极其短暂.

}

}  

* atomic.LoadInt32接受一个*int32类型的指针值

* 返回该指针指向的那个值

存储Store

与读取操作相对应的是写入操作。 而sync/atomic包也提供了与原子的载入函数相对应的原子的值存储函数。 以Store为前缀

• 在原子地存储某个值的过程中，任何CPU都不会进行针对同一个值的读或写操作。
• 原子的值存储操作总会成功，因为它并不会关心被操作值的旧值是什么
• 和CAS操作有着明显的区别

 fmt.Println("======Store value=======")

atomic.StoreInt32(&value, 10)

fmt.Println(value)  

交换Swap

• 与CAS操作不同，原子交换操作不会关心被操作的旧值。
• 它会直接设置新值
• 它会返回被操作值的旧值
• 此类操作比CAS操作的约束更少，同时又比原子载入操作的功能更强

实际案例(继续改造上一版代码)

 //数据文件的实现类型

type myDataFile struct {

f *os.File //文件

fmutex sync.RWMutex //被用于文件的读写锁

rcond *sync.Cond //读操作需要用到的条件变量

woffset int64 // 写操作需要用到的偏移量

roffset int64 // 读操作需要用到的偏移量

dataLen uint32 //数据块长度

}

//此处省略...

func (df *myDataFile) Read() (rsn int64, d Data, err error){

// 读取并更新读偏移量

var offset int64

for {

offset = atomic.LoadInt64(&df.roffset)

if atomic.CompareAndSwapInt64(&df.roffset, offset, (offset + int64(df.dataLen))){

break

}

}

//读取一个数据块,最后读取的数据块序列号

rsn = offset / int64(df.dataLen)

bytes := make([]byte, df.dataLen)

//读写锁:读锁定

df.fmutex.RLock()

defer df.fmutex.RUnlock()

for {

_, err = df.f.ReadAt(bytes, offset)

if err != nil {

if err == io.EOF {

//暂时放弃fmutex的 读锁,并等待通知的到来

df.rcond.Wait()

continue

}

}

break

}

d = bytes

return

}

func (df *myDataFile) Write(d Data) (wsn int64, err error){

//读取并更新写的偏移量

var offset int64

for {

offset = atomic.LoadInt64(&df.woffset)

if atomic.CompareAndSwapInt64(&df.woffset, offset, (offset + int64(df.dataLen))){

break

}

}

//写入一个数据块,最后写入数据块的序号

wsn = offset / int64(df.dataLen)

var bytes []byte

if len(d) > int(df.dataLen){

bytes = d[0:df.dataLen]

}else{

bytes = d

}

df.fmutex.Lock()

defer df.fmutex.Unlock()

_, err = df.f.Write(bytes)

//发送通知

df.rcond.Signal()

return

}

func (df *myDataFile) Rsn() int64{

offset := atomic.LoadInt64(&df.roffset)

return offset / int64(df.dataLen)

}

func (df *myDataFile) Wsn() int64{

offset := atomic.LoadInt64(&df.woffset)

return offset / int64(df.dataLen)

}  

完整代码放在GITHUB：