Go并发原子操作 waitGroup 对象池

原子操作:

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

Go语言提供的原子操作都是非侵入式的.它们由标准库代码包sync/atomic中的众多函数代表.可以通过调用这些函数对几种简单类型执行原子操作.

增或减:

用于增或减的原子操作(以下简称原子增或减操作)的函数名称都以Add为前缀.后跟针对具体类型的名称.

示例:

func main() {
    var a int32 = 1
    var b int64 = 1
    int32Value := atomic.AddInt32(&a, 1)
    int64Value := atomic.AddInt64(&b, 1)
    fmt.Println(int32Value)
    fmt.Println(int64Value)
    reduceInt32 := atomic.AddInt32(&a, -1)
    reduceInt64 := atomic.AddInt64(&b, -1)
    fmt.Println(reduceInt32)
    fmt.Println(reduceInt64)
}

执行结果:

比较并交换:

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

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

函数接受三个参数.第一个参数的值是指向被操作的指针值.该值的类型是*int32.后两个参数的类型也是32.分别代表被操作的旧值和新值.函数在被调用后.会先判断参数addr的指向和旧值old的值是否相等.当判断成功后.才会用新值替换旧值.否则.后面的替换就会被忽略.

与前面的锁相比.CAS操作有明显不同.它总是假设被操作值未曾改变(即与旧值相等).并一旦确认这个假设的真实性就立即进行值替换.使用锁则是更加谨慎的做法.总是先假设会有并发操作会修改被操作值.并需要使用锁将相关操作放入临界区加以保护.可以说.使用锁的做法趋于悲观.而CAS操作的做法比较乐观.

CAS操作的优势是.可以在不创建互斥量和不形成临界区的情况下.完成并发安全的值替换操作.可以大大减少同步对程序性能的损耗.当然,CAS也有劣势.是被操作值被频繁变更的情况下.CAS操作并不那么容易成功.有时候.可能不得不利用for循环来进行多次尝试.

示例:

func main() {
    var value int32
    for i := 0; i < 10; i++ {
       go func() {
          for {
             v := value
             if atomic.CompareAndSwapInt32(&v, 0, 1) {
                break
             }
          }
       }()
    }
}

从上面例子可以看出.只有保证CAS操作成功后.for循环才会结束退出循环.CAS操作虽然不会阻塞goroutine.但是会不断自旋造成性能上的消耗.

载入:

前面展示的for循环中.使用语句v:=value为变量v赋值.要注意.在读取value的过程中.并不能保证没有对此值的并发读写操作.为了原子的读取某个值.sync/atomic代码包也提供了一系列函数.这些函数的名称都以"load"(意为载入)为前缀.

示例:

func main() {
    var value int32
    for i := 0; i < 10; i++ {
       go func() {
          for {
             v := atomic.LoadInt32(&value)
             if atomic.CompareAndSwapInt32(&v, 0, 1) {
                break
             }
          }
       }()
    }
}

函数atomic.LoadInt32接受一个*int32类型的指针值.并会返回该指针值指向的那个值.

注意:虽然这里使用了atomic.LoadInt32函数原子的载入value的值.后面的CAS操作仍然是有必要的.因为.赋值语句后面的if语句并不会原子执行.在它们执行期间.CPU仍然可能执行其它针对Value的操作.

存储:

与读操作相对应的是写入操作.而sync/atomic包也提供了对应的存储函数.这些函数的名称均以"Store"为前缀.

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

原子的值存储操作总会成功.因为它并不关系操作的旧值是什么.这与前面的CAS操作有明显区别.

函数atomic.StoreInt32会接受两个参数.第一个参数的类型是*int32.它同样是指向被操作值的指针值.第二个参数是int32类型.它的值是欲存储的新值.

交换:

在sync/atomic代码包中还有一些函数.它们的功能与前文所讲的CAS操作和原子载入操作都有相似之处.这里的功能可以称为原子交换操作.这类函数的名称都以"Swap"为前缀.

与CAS操作不同.原子操作不会关心被操作的旧值.而是直接设置新值.但它又比原子存储操作多做了一步.它会返回被操作的旧值.此类操作比CAS操作的约束更少.同时又比原子载入操作的功能更强.

func SwapInt32(addr *int32, new int32) (old int32)

从函数可知.它接受两个参数.其中第一个参数代表了被操作的内存地址的*int32类型值.第二个参数用来表示新值.注意.该函数是有结果值的.该值即被新值替换掉的旧值.函数被调用后.会把第二个参数的值置于第一个参数值所代表的的内存地址上.并将之前在该地址上的那个值作为结果返回.

原子值:

sync/atomic.Value是一个结构体类型.暂且称为原子类型.它用于存储需要原子读写的值.与sync/atomic包中的其它函数不同.sync/atomic.Value可接受的被操作值的类型不限.

该类型有两个公开的指针方法Load和Store.前者用于原子的读取原子值实例中存储的值.它会返回一个interface{}类型的结果且不接受任何参数.后者用于原子的在原子实例中存储一个值.它接受一个interface{}类型的参数而没有任何结果.在未曾通过Store方法向原子值实例存储值之前.它的Load方法总会返回nil.

对于原子值实例的Store方法有两个限制.第一.作为参数传入该方法的值不能为nil.第二.作为参数传入该方法的值必须与之前传入的值(如果有的话)的类型相同.一旦原子值实例存储了某一个类型的值.那么它之后存储的值就必须是该类型的.如果违反了任意一个限制.对该方法的调用都会引发一个运行是恐慌.

严格来说.sync/atomic.Value类型的变量一旦声明.其值就不应该被复制到它处.作为源值赋给别的变量 作为参数值传入参数 作为结果值从函数返回 作为元素值通过通道传递等都会造成值的复制.所以这类变量之上不应该实施这些操作.虽然编译不会错误.但Go标准工具go vet会报告此类不正确(或说有安全隐患).不过原子类型不会有这个问题.根本原因.对结构体值的复制不但会生成该值的副本.还会生成其中字段的副本.这样一来.本应施加于此的并发安全也就失效了.向副本存储值的操作也与原值无关了.

示例:

func main() {
    var countVal atomic.Value
    countVal.Store([]int{1, 3, 5, 7})
    anotherStore(countVal)
    fmt.Printf("The count value: %v\n", countVal.Load())
}

func anotherStore(countVal atomic.Value) {
    countVal.Store([]int{2, 4, 6, 8})
}

执行结果:

使用场景:

type ConcurrentArray interface {
    //设置指定索引上的元素值.
    Set(index uint32,elem int) (err error)
    //用于获取指定索引元素的值.
    Get(index uint32) (elem int, err error)
    //用于获取元素的长度.
    Len() uint32
}

接口实现:

type concurrentArray struct {
    length uint32
    val atomic.Value
}

创建函数:

func NewConcurrentArray(length uint32) ConcurrentArray {
    array := concurrentArray{}
    array.length = length
    array.val.Store(make([]int, array.length))
    return &array
}

接口实现方法:

func (array *concurrentArray) Len() uint32 {
    return array.length
}

func (array *concurrentArray) Set(index uint32, elem int) (err error) {
    newArray := make([]int, array.length)
    copy(newArray, array.val.Load().([]int))
    newArray[index] = elem
    array.val.Store(newArray)
    return
}

func (array *concurrentArray) Get(index uint32) (elem int, err error) {
    elem = array.val.Load().([]int)[index]
    return elem,nil
}

只执行一次:

与互斥锁和读写锁一样,sync.Once也是开箱即用.

var once sync.Once

once.Do(func() {fmt.Println("Once")})

声明了一个名为once的sync.Once类型的变量.然后用执行它的Do方法.Do方法接受一个无参数 无结果的函数值作为参数.该方法一旦被调用.就会调用作为参数的那个函数.对同一个sync.Once类型值的Do方法的有效调用次数永远是一次.

示例:

func main() {
    var count int
    var once sync.Once
    max := rand.Intn(100)
    for i := 0; i < max; i++ {
       once.Do(func() {
          count++
       })
    }
    fmt.Printf("Count: %d\n", count)
}

waitGroup:

sync.WAitGroup类型的值是并发安全的.也是开箱即用的.在声明中var wg sync.WaitGroup之后.就可以直接使用wg变量了.该类型有三个方法.即Add Done和wait.

func main() {
    var wg sync.WaitGroup
    wg.Add(3)
    for i := 0; i < 3; i++ {
       go func() {
          fmt.Printf("协程:%d执行了.\n", i)
          time.Sleep(1 * time.Second)
          wg.Done()
       }()
    }
    fmt.Println("等待协程开始执行")
    wg.Wait()
    fmt.Println("所有协程都执行完了.结束主线程")
}

使用规则:

1).对一个sync.WaitGroup类型值的Add方法的第一次调用.发生在调用该值的Done之前.

2).对一个sync.WaitGroup类型值的Add方法的第一次调用.同样发生在调用该值的wait方法之前.

3).在一个sync.WaitGroup类型值的生命周期内.其中的给定计数总是由起初的0变为某个正整数(或先后变为某几个正整数).然后在回归为0.完成这样的变化曲线所用的时间称为一个计数周期.

4).给定计数的每次变化都是由对Add方法或Done方法的调用引起的.一个计数的周期总是从Add方法调用开始的.并且也总是以对Add方法或Done方法的调用为结束标志.在一个计数周期之内调用wait方法.就会使调用所在的goroutine阻塞.直到该计数周期结束的那一刻.

5).sync.WaitGroup类型值是可以复用的.此类值的生命周期可以包含任意个计数周期.一旦一个计数周期结束.在前面对该值的方法调用所产生的作用就会消失.它们不会影响该值的后续计数周期.一个sync.WaitGroup类型值在其每个计数周期中的状态和作用都是独立的.

临时对象池:

sync.Pool类型值可以看作存放临时值的容器.此类容器是自动伸缩的 高效的.同时也是并发安全的.为了描述方便.把sync.Pool类型的值称为"临时对象池".而把存于其中的值称为"对象值".

在用复合字面量初始化一个临时对象池的时候.可以为它唯一的公开字段New赋值.该字段的类型是func() interface{}.即一个函数类型.赋给该字段的函数会被临时对象池用来创建对象值.不过.该函数一般仅在池中无可用对象值的时候才被调用..把这个函数称为"对象值生成函数".

sync.Pool类型由两个公开的指针方法:Get和Put.前者的功能是从池中获取一个interface{}类型的值.后者的作用则是把一个interface{}类型的值放置于池中.

通过Get方法获取到的值是任意的.如果一个临时对象池的Put方法从未被调用过.且它的New字段也未曾被赋予一个非null的函数值.那么它的Get方法返回的结果就一定会是nil.临时对象池在功能上与一个通用的缓存池有一些相似.实际上.临时对象池本身的特性决定了它是一个很独特的同步工具.

特性:

1).临时对象池可以把由其中的对象值产生的存储压力进行分摊.更进一步说.它会专门为每一个与操作它的goroutine相关联的P建立本地池.在临时对象池的Get方法被调用时.它一般会先尝试从与本地P对应的那个本地私有池和本地共享池中获取一个对象池.如果获取失败.它就会尝试从其它P的本地共享池中偷一个对象值并直接返回给调用方.如果依然未果.它就只能把希望寄托于当前临时对象池的对象生成函数了.注意.这个对象值生成函数生成的对象永远不会被放置到池中.而是会被直接返回给调用方.另一方面.临时对象池的Put方法会把它的参数值存放到本地P的本地池中.每个相关的P的本地共享池中所有对象值.都是在当前临时对象池的范围内共享的.也就是说.它们随时可能被偷走.

2).对垃圾回收友好.垃圾回收的执行一般会使临时对象池中的对象值全部移除.也就是说.即使永远不会显示的从临时对象池中取走某个对象值.该对象值也不会永远待在临时对象池中.它的生命周期取决于垃圾回收任务的下一次执行时间.

func main() {
    // 1. 保存原始GC配置，函数结束后恢复（正确写法）
    originalGC := debug.SetGCPercent(-1)
    defer debug.SetGCPercent(originalGC)

    var count int32
    // 统一类型：全部存储 int32 类型
    newFunc := func() interface{} {
       return atomic.AddInt32(&count, 1)
    }
    pool := sync.Pool{New: newFunc}

    // 池空，调用New函数
    v1 := pool.Get()
    fmt.Printf("Value 1: %v\n", v1)

    // 存入与New函数返回值 相同类型 的对象（int32）
    pool.Put(int32(10))
    pool.Put(int32(11))
    pool.Put(int32(12))
    v2 := pool.Get()
    fmt.Printf("Value 2: %v\n", v2)

    // 手动触发GC，Pool缓存会被回收
    debug.SetGCPercent(100)
    runtime.GC()

    // GC后缓存清空，再次调用New
    v3 := pool.Get()
    fmt.Printf("Value 3: %v\n", v3)

    // 清空New函数
    pool.New = nil
    v4 := pool.Get()
    fmt.Printf("Value 4: %v\n", v4)
}

到此这篇关于Go并发原子操作 waitGroup 对象池的文章就介绍到这了,更多相关Go原子操作waitGroup 对象池内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！

最新评论